danvics/Pulse
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add container state filtering to Docker agent

Browse source
This commit is contained in:
rcourtman 2025-10-25 21:40:59 +00:00
parent 535421b8ea
commit 8e83eaf823
4 changed files with 184 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
cmd/pulse-docker-agent/main.go
View file

@ -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
}

5
docs/DOCKER_MONITORING.md
View file

@ -84,6 +84,10 @@ You can also repeat `--target https://pulse.example.com|<token>` 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` |

137
internal/dockeragent/agent.go
View file

@ -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)

23
internal/dockeragent/agent_internal_test.go
View file

@ -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")
}
}