diff --git a/agent/internal/grpc/client.go b/agent/internal/grpc/client.go index fa6daff..7ad9878 100644 --- a/agent/internal/grpc/client.go +++ b/agent/internal/grpc/client.go @@ -80,7 +80,11 @@ func (c *Client) Register(serverID, preRegToken, hostname, ipAddress, osInfo str return resp.AgentToken, nil } -func (c *Client) SyncKeys(serverID, agentToken, version string) ([]string, error) { +// SyncKeys returns the whole response rather than just the keys: the poll now +// also carries CollectPackages, and a second RPC purely to learn one boolean +// would be a message every 30 seconds for a value that changes at most when a +// licence does. +func (c *Client) SyncKeys(serverID, agentToken, version string) (*pb.SyncResponse, error) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -92,7 +96,21 @@ func (c *Client) SyncKeys(serverID, agentToken, version string) ([]string, error if err != nil { return nil, err } - return resp.PublicKeys, nil + return resp, nil +} + +// ReportPackages sends a package report and returns whether the server wants +// the full list. Given a longer deadline than the other unary calls because the +// full body is ~150KB on a slow link. +func (c *Client) ReportPackages(req *pb.ReportPackagesRequest) (bool, error) { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + resp, err := c.client.ReportPackages(ctx, req) + if err != nil { + return false, err + } + return resp.NeedFull, nil } func (c *Client) UploadGeneratedKey(serverID, agentToken, publicKey, privateKey, label string) (string, error) { diff --git a/agent/internal/sync/packages.go b/agent/internal/sync/packages.go new file mode 100644 index 0000000..6ebb316 --- /dev/null +++ b/agent/internal/sync/packages.go @@ -0,0 +1,90 @@ +package agentsync + +import ( + "log" + "runtime" + "sync/atomic" + + "gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/config" + grpcclient "gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/grpc" + "gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/grpc/pb" + "gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/packages" +) + +// collectPackagesFlag is written by the 30s key poll and read by the hourly +// package loop — two different goroutines, hence the atomic. +// +// It defaults to false, so an agent that has not yet completed a poll, or is +// talking to a server too old to send the field, collects nothing. Off is the +// safe default: collecting without a licence costs the customer storage they +// are not paying for. +var collectPackagesFlag atomic.Bool + +func collectPackagesEnabled() bool { return collectPackagesFlag.Load() } + +// reportPackages offers a hash of the installed package set and sends the full +// list only if the server does not already hold it. +// +// It runs on the same hourly cadence as the update check because a package set +// changes on roughly the same schedule, and reusing that loop means one timer +// rather than two. +func reportPackages(client *grpcclient.Client, cfg *config.Config) { + if runtime.GOOS != "linux" { + return + } + if !collectPackagesEnabled() { + return + } + + osrel, pkgs, err := packages.Collect() + if err != nil { + log.Printf("package collection error: %v", err) + return + } + + pbOS := pb.OSRelease{ + Family: osrel.Family, + VersionId: osrel.VersionID, + Arch: osrel.Arch, + } + hash := packages.Hash(pkgs) + + // The offer: hash only, no body. On an unchanged host this is the whole + // exchange, which is the point of the handshake. + needFull, err := client.ReportPackages(&pb.ReportPackagesRequest{ + ServerId: cfg.ServerID, + AgentToken: cfg.AgentToken, + Hash: hash, + Os: pbOS, + }) + if err != nil { + log.Printf("ReportPackages offer error: %v", err) + return + } + if !needFull { + return + } + + pbPkgs := make([]pb.InstalledPackage, len(pkgs)) + for i, p := range pkgs { + pbPkgs[i] = pb.InstalledPackage{ + Name: p.Name, + Version: p.Version, + Epoch: int32(p.Epoch), + Arch: p.Arch, + SourceName: p.SourceName, + } + } + + if _, err := client.ReportPackages(&pb.ReportPackagesRequest{ + ServerId: cfg.ServerID, + AgentToken: cfg.AgentToken, + Hash: hash, + Os: pbOS, + Packages: pbPkgs, + }); err != nil { + log.Printf("ReportPackages full error: %v", err) + return + } + log.Printf("reported %d installed packages", len(pkgs)) +} diff --git a/agent/internal/sync/sync.go b/agent/internal/sync/sync.go index 44b845f..538c008 100644 --- a/agent/internal/sync/sync.go +++ b/agent/internal/sync/sync.go @@ -92,11 +92,18 @@ func Run(ctx context.Context, cfg *config.Config, version string) error { } func poll(client *grpcclient.Client, cfg *config.Config, version string) error { - desired, err := client.SyncKeys(cfg.ServerID, cfg.AgentToken, version) + resp, err := client.SyncKeys(cfg.ServerID, cfg.AgentToken, version) if err != nil { return fmt.Errorf("SyncKeys: %w", err) } + // Stored atomically: the hourly package loop reads this from another + // goroutine. Absent on the wire decodes as false, so an older server leaves + // collection off rather than on. + collectPackagesFlag.Store(resp.CollectPackages) + + desired := resp.PublicKeys + if runtime.GOOS != "linux" { return nil } @@ -395,6 +402,11 @@ func runUpdateCheck(ctx context.Context, cfg *config.Config) { return } log.Printf("reported %d available OS updates", len(pkgs)) + + // Same hourly cadence, same connection. A package set changes on + // roughly the schedule available updates do, so this needs no timer of + // its own. + reportPackages(client, cfg) } doCheck()