diff --git a/cmd/pulse-docker-agent/main.go b/cmd/pulse-docker-agent/main.go index 7718e38..cb067ef 100644 --- a/cmd/pulse-docker-agent/main.go +++ b/cmd/pulse-docker-agent/main.go @@ -14,17 +14,24 @@ import ( "github.com/rs/zerolog" ) -type targetFlagList []string +type stringFlagList []string -func (l *targetFlagList) String() string { +func (l *stringFlagList) String() string { return strings.Join(*l, ",") } -func (l *targetFlagList) Set(value string) error { +func (l *stringFlagList) Set(value string) error { *l = append(*l, value) return nil } +func (l stringFlagList) Values() []string { + if len(l) == 0 { + return nil + } + return append([]string(nil), l...) +} + func main() { cfg := loadConfig() @@ -58,6 +65,7 @@ func loadConfig() dockeragent.Config { envInsecure := strings.TrimSpace(os.Getenv("PULSE_INSECURE_SKIP_VERIFY")) envNoAutoUpdate := strings.TrimSpace(os.Getenv("PULSE_NO_AUTO_UPDATE")) envTargets := strings.TrimSpace(os.Getenv("PULSE_TARGETS")) + envContainerStates := strings.TrimSpace(os.Getenv("PULSE_CONTAINER_STATES")) defaultInterval := 30 * time.Second if envInterval != "" { @@ -73,8 +81,10 @@ func loadConfig() dockeragent.Config { agentIDFlag := flag.String("agent-id", envAgentID, "Override agent identifier") insecureFlag := flag.Bool("insecure", parseBool(envInsecure), "Skip TLS certificate verification") noAutoUpdateFlag := flag.Bool("no-auto-update", parseBool(envNoAutoUpdate), "Disable automatic agent updates") - var targetFlags targetFlagList + var targetFlags stringFlagList flag.Var(&targetFlags, "target", "Pulse target in url|token[|insecure] format. Repeat to send to multiple Pulse instances") + var containerStateFlags stringFlagList + flag.Var(&containerStateFlags, "container-state", "Only include containers whose status matches this value (repeat to allow multiple). Allowed values: created,running,restarting,removing,paused,exited,dead.") flag.Parse() @@ -86,7 +96,7 @@ func loadConfig() dockeragent.Config { targets := make([]dockeragent.TargetConfig, 0) if len(targetFlags) > 0 { - parsedTargets, err := parseTargetSpecs(targetFlags) + parsedTargets, err := parseTargetSpecs(targetFlags.Values()) if err != nil { fmt.Fprintf(os.Stderr, "error: %v\n", err) os.Exit(1) @@ -118,6 +128,14 @@ func loadConfig() dockeragent.Config { interval = 30 * time.Second } + containerStates := make([]string, 0) + if len(containerStateFlags) > 0 { + containerStates = append(containerStates, containerStateFlags.Values()...) + } + if envContainerStates != "" { + containerStates = append(containerStates, splitStringList(envContainerStates)...) + } + return dockeragent.Config{ PulseURL: pulseURL, APIToken: token, @@ -127,6 +145,7 @@ func loadConfig() dockeragent.Config { InsecureSkipVerify: *insecureFlag, DisableAutoUpdate: *noAutoUpdateFlag, Targets: targets, + ContainerStates: containerStates, } } @@ -204,3 +223,27 @@ func splitTargetSpecs(value string) []string { } return result } + +func splitStringList(value string) []string { + if value == "" { + return nil + } + + items := strings.FieldsFunc(value, func(r rune) bool { + switch r { + case ',', ';', '\n', '\r': + return true + default: + return false + } + }) + + result := make([]string, 0, len(items)) + for _, item := range items { + if trimmed := strings.TrimSpace(item); trimmed != "" { + result = append(result, trimmed) + } + } + + return result +} diff --git a/docs/DOCKER_MONITORING.md b/docs/DOCKER_MONITORING.md index d9cf264..921808e 100644 --- a/docs/DOCKER_MONITORING.md +++ b/docs/DOCKER_MONITORING.md @@ -84,6 +84,10 @@ You can also repeat `--target https://pulse.example.com|` on the command The binary reads standard Docker environment variables. If you already use TLS-secured remote sockets set `DOCKER_HOST`, `DOCKER_TLS_VERIFY`, etc. as normal. To skip TLS verification for Pulse (not recommended) add `--insecure` or `PULSE_INSECURE_SKIP_VERIFY=true`. +### Filtering container states + +High churn environments can flood Pulse with noise from short-lived tasks. Restrict the agent to the container states you care about by repeating `--container-state` (for example, `--container-state running --container-state paused`) or by exporting `PULSE_CONTAINER_STATES=running,paused`. Allowed values match Docker’s status filter: `created`, `running`, `restarting`, `removing`, `paused`, `exited`, and `dead`. If no values are provided the agent reports every container, mirroring the previous behaviour. + ### Multiple Pulse instances A single `pulse-docker-agent` process can now serve any number of Pulse backends. Each target entry keeps its own API token and TLS preference, and Pulse de-duplicates reports using the shared agent ID / machine ID. This avoids running duplicate agents on busy Docker hosts. @@ -134,6 +138,7 @@ docker run -d \ | `--token`, `PULSE_TOKEN`| Pulse API token with `docker:report` scope (required). | — | | `--target`, `PULSE_TARGETS` | One or more `url|token[|insecure]` entries to fan-out reports to multiple Pulse servers. Separate entries with `;` or repeat the flag. | — | | `--interval`, `PULSE_INTERVAL` | Reporting cadence (supports `30s`, `1m`, etc.). | `30s` | +| `--container-state`, `PULSE_CONTAINER_STATES` | Limit reports to specific Docker statuses (`created`, `running`, `restarting`, `removing`, `paused`, `exited`, `dead`). Separate multiple values with commas/semicolons or repeat the flag. | — | | `--hostname`, `PULSE_HOSTNAME` | Override host name reported to Pulse. | Docker info / OS hostname | | `--agent-id`, `PULSE_AGENT_ID` | Stable ID for the agent (useful for clustering). | Docker engine ID / machine-id | | `--insecure`, `PULSE_INSECURE_SKIP_VERIFY` | Skip TLS cert validation (unsafe). | `false` | diff --git a/internal/dockeragent/agent.go b/internal/dockeragent/agent.go index 237bd1e..d9bea6a 100644 --- a/internal/dockeragent/agent.go +++ b/internal/dockeragent/agent.go @@ -19,6 +19,7 @@ import ( "github.com/docker/docker/api/types" containertypes "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" "github.com/docker/docker/client" agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker" "github.com/rs/zerolog" @@ -41,20 +42,34 @@ type Config struct { InsecureSkipVerify bool DisableAutoUpdate bool Targets []TargetConfig + ContainerStates []string Logger *zerolog.Logger } +var allowedContainerStates = map[string]string{ + "created": "created", + "restarting": "restarting", + "running": "running", + "removing": "removing", + "paused": "paused", + "exited": "exited", + "dead": "dead", + "stopped": "exited", +} + // Agent collects Docker metrics and posts them to Pulse. type Agent struct { - cfg Config - docker *client.Client - httpClients map[bool]*http.Client - logger zerolog.Logger - machineID string - hostName string - cpuCount int - targets []TargetConfig - hostID string + cfg Config + docker *client.Client + httpClients map[bool]*http.Client + logger zerolog.Logger + machineID string + hostName string + cpuCount int + targets []TargetConfig + allowedStates map[string]struct{} + stateFilters []string + hostID string } // ErrStopRequested indicates the agent should terminate gracefully after acknowledging a stop command. @@ -89,6 +104,12 @@ func New(cfg Config) (*Agent, error) { cfg.APIToken = targets[0].Token cfg.InsecureSkipVerify = targets[0].InsecureSkipVerify + stateFilters, err := normalizeContainerStates(cfg.ContainerStates) + if err != nil { + return nil, err + } + cfg.ContainerStates = stateFilters + dockerClient, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) if err != nil { return nil, fmt.Errorf("failed to create docker client: %w", err) @@ -130,13 +151,19 @@ func New(cfg Config) (*Agent, error) { } agent := &Agent{ - cfg: cfg, - docker: dockerClient, - httpClients: httpClients, - logger: *logger, - machineID: machineID, - hostName: hostName, - targets: cfg.Targets, + cfg: cfg, + docker: dockerClient, + httpClients: httpClients, + logger: *logger, + machineID: machineID, + hostName: hostName, + targets: cfg.Targets, + allowedStates: make(map[string]struct{}, len(stateFilters)), + stateFilters: stateFilters, + } + + for _, state := range stateFilters { + agent.allowedStates[state] = struct{}{} } return agent, nil @@ -181,6 +208,36 @@ func normalizeTargets(raw []TargetConfig) ([]TargetConfig, error) { return normalized, nil } +func normalizeContainerStates(raw []string) ([]string, error) { + if len(raw) == 0 { + return nil, nil + } + + normalized := make([]string, 0, len(raw)) + seen := make(map[string]struct{}, len(raw)) + + for _, value := range raw { + state := strings.ToLower(strings.TrimSpace(value)) + if state == "" { + continue + } + + canonical, ok := allowedContainerStates[state] + if !ok { + return nil, fmt.Errorf("unsupported container state %q", value) + } + + if _, exists := seen[canonical]; exists { + continue + } + + seen[canonical] = struct{}{} + normalized = append(normalized, canonical) + } + + return normalized, nil +} + // Run starts the collection loop until the context is cancelled. func (a *Agent) Run(ctx context.Context) error { interval := a.cfg.Interval @@ -294,13 +351,28 @@ func (a *Agent) buildReport(ctx context.Context) (agentsdocker.Report, error) { } func (a *Agent) collectContainers(ctx context.Context) ([]agentsdocker.Container, error) { - list, err := a.docker.ContainerList(ctx, containertypes.ListOptions{All: true}) + options := containertypes.ListOptions{All: true} + if len(a.stateFilters) > 0 { + filterArgs := filters.NewArgs() + for _, state := range a.stateFilters { + filterArgs.Add("status", state) + } + options.Filters = filterArgs + } + + list, err := a.docker.ContainerList(ctx, options) if err != nil { return nil, fmt.Errorf("failed to list containers: %w", err) } containers := make([]agentsdocker.Container, 0, len(list)) for _, summary := range list { + if len(a.allowedStates) > 0 { + if _, ok := a.allowedStates[strings.ToLower(summary.State)]; !ok { + continue + } + } + container, err := a.collectContainer(ctx, summary) if err != nil { a.logger.Warn().Str("container", strings.Join(summary.Names, ",")).Err(err).Msg("Failed to collect container stats") @@ -322,19 +394,28 @@ func (a *Agent) collectContainer(ctx context.Context, summary types.Container) ( return agentsdocker.Container{}, fmt.Errorf("inspect: %w", err) } - statsResp, err := a.docker.ContainerStatsOneShot(containerCtx, summary.ID) - if err != nil { - return agentsdocker.Container{}, fmt.Errorf("stats: %w", err) - } - defer statsResp.Body.Close() + var ( + cpuPercent float64 + memUsage int64 + memLimit int64 + memPercent float64 + ) - var stats containertypes.StatsResponse - if err := json.NewDecoder(statsResp.Body).Decode(&stats); err != nil { - return agentsdocker.Container{}, fmt.Errorf("decode stats: %w", err) - } + if inspect.State.Running || inspect.State.Paused { + statsResp, err := a.docker.ContainerStatsOneShot(containerCtx, summary.ID) + if err != nil { + return agentsdocker.Container{}, fmt.Errorf("stats: %w", err) + } + defer statsResp.Body.Close() - cpuPercent := calculateCPUPercent(stats, a.cpuCount) - memUsage, memLimit, memPercent := calculateMemoryUsage(stats) + var stats containertypes.StatsResponse + if err := json.NewDecoder(statsResp.Body).Decode(&stats); err != nil { + return agentsdocker.Container{}, fmt.Errorf("decode stats: %w", err) + } + + cpuPercent = calculateCPUPercent(stats, a.cpuCount) + memUsage, memLimit, memPercent = calculateMemoryUsage(stats) + } createdAt := time.Unix(summary.Created, 0) diff --git a/internal/dockeragent/agent_internal_test.go b/internal/dockeragent/agent_internal_test.go index 41ad97a..f32fc4b 100644 --- a/internal/dockeragent/agent_internal_test.go +++ b/internal/dockeragent/agent_internal_test.go @@ -1,6 +1,9 @@ package dockeragent -import "testing" +import ( + "reflect" + "testing" +) func TestNormalizeTargets(t *testing.T) { targets, err := normalizeTargets([]TargetConfig{ @@ -33,3 +36,21 @@ func TestNormalizeTargetsInvalid(t *testing.T) { t.Fatalf("expected error for missing token") } } + +func TestNormalizeContainerStates(t *testing.T) { + states, err := normalizeContainerStates([]string{"running", "Exited", "running", "stopped"}) + if err != nil { + t.Fatalf("normalizeContainerStates returned error: %v", err) + } + + expected := []string{"running", "exited"} + if !reflect.DeepEqual(states, expected) { + t.Fatalf("expected %v, got %v", expected, states) + } +} + +func TestNormalizeContainerStatesInvalid(t *testing.T) { + if _, err := normalizeContainerStates([]string{"unknown"}); err == nil { + t.Fatalf("expected error for invalid container state") + } +}