134 lines
3.7 KiB
Go
134 lines
3.7 KiB
Go
package hub
|
|
|
|
import (
|
|
"context"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/shcizo/package-updater/internal/master/events"
|
|
"github.com/shcizo/package-updater/internal/master/model"
|
|
"github.com/shcizo/package-updater/internal/protocol"
|
|
)
|
|
|
|
// StartJob sends job.start over exactly the connection r names. It returns
|
|
// model.ErrAgentOffline if that connection is gone, was replaced by a newer
|
|
// one, or has not finished its hello.
|
|
func (h *Hub) StartJob(r model.Route, jobID int64, start protocol.JobStart) error {
|
|
h.mu.Lock()
|
|
c := h.conns[r.AgentID]
|
|
usable := c != nil && c.id == r.ConnID && c.hello != nil
|
|
h.mu.Unlock()
|
|
if !usable {
|
|
return model.ErrAgentOffline
|
|
}
|
|
env, err := protocol.EncodeJob(protocol.TypeJobStart, strconv.FormatInt(jobID, 10), start)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// A job whose write later fails stays queued; the agent's next hello
|
|
// settles it as lost.
|
|
if !c.send(env) {
|
|
return model.ErrAgentOffline
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// parseJobIDs converts the agent's job IDs to integers, calling skip for each
|
|
// one that is not a number.
|
|
func parseJobIDs(ids []string, skip func(id string)) []int64 {
|
|
out := make([]int64, 0, len(ids))
|
|
for _, s := range ids {
|
|
id, err := strconv.ParseInt(s, 10, 64)
|
|
if err != nil {
|
|
skip(s)
|
|
continue
|
|
}
|
|
out = append(out, id)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// cleanOutput drops NUL bytes, which Postgres text columns reject. Without
|
|
// this a result containing one could never be stored and would be resent
|
|
// forever.
|
|
func cleanOutput(s string) string { return strings.ReplaceAll(s, "\x00", "") }
|
|
|
|
func (s *session) handleJobProgress(ctx context.Context, env protocol.Envelope) error {
|
|
h := s.hub
|
|
v, err := protocol.Decode[protocol.JobProgress](env)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id, err := strconv.ParseInt(env.ID, 10, 64)
|
|
if err != nil {
|
|
h.log.Warn("job_progress_ignored", "agent", s.agent.Name, "job_id", env.ID, "reason", "not a number")
|
|
return nil
|
|
}
|
|
applied, err := h.store.JobProgress(ctx, s.agent.ID, id, cleanOutput(v.Output), h.now())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if applied {
|
|
h.pub.Publish(events.Jobs)
|
|
h.pub.Publish(events.JobTopic(id))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *session) handleJobResult(ctx context.Context, env protocol.Envelope) error {
|
|
h := s.hub
|
|
v, err := protocol.Decode[protocol.JobResult](env)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ack, err := protocol.EncodeJob(protocol.TypeJobAck, env.ID, protocol.JobAck{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id, err := strconv.ParseInt(env.ID, 10, 64)
|
|
if err != nil {
|
|
h.log.Warn("job_result_ignored", "agent", s.agent.Name, "job_id", env.ID, "reason", "not a number")
|
|
s.conn.send(ack)
|
|
return nil
|
|
}
|
|
|
|
fin := model.JobFinish{
|
|
Status: model.JobStatus(v.Status),
|
|
Ref: v.Ref,
|
|
Error: v.Error,
|
|
Output: cleanOutput(v.Output),
|
|
OutputTruncated: v.OutputTruncated,
|
|
StartedAt: v.StartedAt,
|
|
FinishedAt: v.FinishedAt,
|
|
}
|
|
if v.Status != protocol.ResultSucceeded && v.Status != protocol.ResultFailed {
|
|
fin.Status = model.JobFailed
|
|
fin.Ref = ""
|
|
fin.Error = "agent sent invalid status " + strconv.Quote(v.Status)
|
|
}
|
|
if fin.StartedAt.IsZero() {
|
|
fin.StartedAt = h.now()
|
|
}
|
|
if fin.FinishedAt.IsZero() {
|
|
fin.FinishedAt = h.now()
|
|
}
|
|
|
|
// A store error closes the connection without an ack, so the agent
|
|
// resends the result after it reconnects.
|
|
applied, err := h.store.FinishJob(ctx, s.agent.ID, id, fin)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if applied {
|
|
h.pub.Publish(events.Jobs)
|
|
h.pub.Publish(events.JobTopic(id))
|
|
if fin.Status == model.JobSucceeded {
|
|
h.pub.Publish(events.Services)
|
|
}
|
|
} else {
|
|
h.log.Warn("job_result_ignored", "agent", s.agent.Name, "job_id", env.ID, "reason", "no such active job for this agent")
|
|
}
|
|
s.conn.send(ack)
|
|
return nil
|
|
}
|