Go Agent: Architecture & Internals

graph TD
    A[main.go] --> B[agent.Agent]

    B --> C[client.Client\nHTTP to VF server]
    B --> D[poller\nConfig diff engine]
    B --> E[supervisor.Supervisor\nVector process manager]
    B --> F[push.Client\nSSE push channel]
    B --> G[tapper.Manager\nLive event tapping]
    B --> H[selfmetrics.Metrics\nAgent self-observability]

    D -- GET /api/agent/config --> C
    C -- POST /api/agent/heartbeat --> C

    E --> E1[Vector process\npipeline-a]
    E --> E2[Vector process\npipeline-b]
    E --> E3[Vector process\npipeline-n]

    E1 -- scrape :8688/metrics --> I[metrics.ScrapePrometheus]
    E2 -- scrape :8690/metrics --> I
    I --> B

    E1 -- tap via API :8689 --> G
    G -- POST /api/agent/tap-events --> C

    F -- SSE stream --> J[VF server push endpoint]
    J -- config_changed\nsample_request\naction --> B

    H -- :9090/metrics --> K[Prometheus scraper]

┌─────────────────────────────────────────────────────────────┐
│                       Run() select loop                      │
│                                                              │
│  ┌──────────────┐  ┌──────────────┐  ┌───────────────────┐  │
│  │  ticker.C    │  │  pushCh      │  │ immediateHeartbeat│  │
│  │  (poll       │  │  (SSE push   │  │ Ch               │  │
│  │   interval)  │  │   messages)  │  │ (debounced 1s)   │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬──────────┘  │
│         │                 │                    │             │
│         ▼                 ▼                    ▼             │
│   pollAndApply()    handlePushMessage()   sendHeartbeat()   │
│   sendHeartbeat()                                            │
└─────────────────────────────────────────────────────────────┘

main()
  │
  ├─ config.Load()            # parse all VF_* env vars
  ├─ agent.New(cfg)           # construct all subsystems
  │    ├─ client.New(url)
  │    ├─ supervisor.New(vectorBin)
  │    ├─ tapper.New(vectorBin)
  │    └─ selfmetrics.New(pipelinesRunningFn)
  │
  └─ agent.Run()
       ├─ loadOrEnroll()      # load node-token from disk or enroll with server
       ├─ (if VF_METRICS_PORT > 0) go metrics.Serve(port)
       ├─ pollAndApply()      # first poll immediately
       ├─ push.Client.Connect() → go routine
       ├─ runLogFlusher()     → go routine
       └─ sendHeartbeat()     # first heartbeat

pollAndApply()
  └─ poller.Poll()
       └─ returns ActionStart for new pipelines
            │
            └─ supervisor.Start(pipelineID, configPath, ...)
                 ├─ allocates metricsPort = basePort + (portSeq*2 - 1)
                 ├─ allocates apiPort     = basePort + (portSeq*2)
                 ├─ writeSidecarConfig()   → <configPath>.vf-metrics.yaml
                 ├─ exec.Command(vectorBin, --config, configPath,
                 │                         --config, sidecarPath)
                 └─ go monitor(info)       # goroutine watches for exit

monitor(info) goroutine
  ├─ time.Sleep(2s)           # startup grace period → sets status RUNNING
  ├─ cmd.Wait()               # blocks until process exits
  └─ if err != nil (crash):
       info.Status = "CRASHED"
       info.restarts++
       backoff = min(1s << (restarts-1), 60s)
       go func() {
           time.Sleep(backoff)
           startProcess(...)   # restart with same ports
       }()

sources:
  vf_internal_metrics:       # Vector's own internal metrics
    type: internal_metrics
  vf_host_metrics:           # CPU, memory, disk, network from the host
    type: host_metrics

sinks:
  vf_metrics_exporter:       # Prometheus exporter for the agent to scrape
    type: prometheus_exporter
    inputs: ["vf_internal_metrics", "vf_host_metrics"]
    address: "127.0.0.1:<metricsPort>"

api:
  enabled: true
  address: "127.0.0.1:<apiPort>"  # Vector API for tap and sampling