diff --git a/.github/workflows/agent.yml b/.github/workflows/agent.yml index 1741b23..80643a4 100644 --- a/.github/workflows/agent.yml +++ b/.github/workflows/agent.yml @@ -30,6 +30,15 @@ jobs: go-version-file: apps/agent/go.mod cache-dependency-path: apps/agent/go.sum + - name: Check formatting + run: | + unformatted=$(gofmt -l .) + if [ -n "$unformatted" ]; then + echo "not gofmt-formatted:" + echo "$unformatted" + exit 1 + fi + - name: Verify go.mod is tidy run: | go mod tidy diff --git a/.github/workflows/cli.yml b/.github/workflows/cli.yml index 8d1c0b6..f832cf9 100644 --- a/.github/workflows/cli.yml +++ b/.github/workflows/cli.yml @@ -26,6 +26,15 @@ jobs: go-version-file: apps/cli/go.mod cache-dependency-path: apps/cli/go.sum + - name: Check formatting + run: | + unformatted=$(gofmt -l .) + if [ -n "$unformatted" ]; then + echo "not gofmt-formatted:" + echo "$unformatted" + exit 1 + fi + - name: Vet run: go vet ./... diff --git a/apps/agent/cmd/agent/main.go b/apps/agent/cmd/agent/main.go index 84aa811..a00ba32 100644 --- a/apps/agent/cmd/agent/main.go +++ b/apps/agent/cmd/agent/main.go @@ -1,37 +1,37 @@ -package main - -import ( - "context" - "flag" - "fmt" - "os" - - "docksight-agent/internal/app" - "docksight-agent/internal/service" - "docksight-agent/internal/version" -) - -func main() { - configPath := flag.String("config", "config.yaml", "path to agent config.yaml") - showVersion := flag.Bool("version", false, "print version and exit") - flag.Parse() - - if *showVersion { - fmt.Println(version.String()) - return - } - - application := app.New(*configPath) - - // service.Run decides how the agent is supervised: directly in a console, - // or dispatched to the Windows Service Control Manager. The same binary - // serves both, detected at runtime rather than selected by a flag. - err := service.Run(func(ctx context.Context) error { - return application.Run(ctx) - }) - - if err != nil { - fmt.Fprintf(os.Stderr, "docksight-agent: %v\n", err) - os.Exit(1) - } -} +package main + +import ( + "context" + "flag" + "fmt" + "os" + + "docksight-agent/internal/app" + "docksight-agent/internal/service" + "docksight-agent/internal/version" +) + +func main() { + configPath := flag.String("config", "config.yaml", "path to agent config.yaml") + showVersion := flag.Bool("version", false, "print version and exit") + flag.Parse() + + if *showVersion { + fmt.Println(version.String()) + return + } + + application := app.New(*configPath) + + // service.Run decides how the agent is supervised: directly in a console, + // or dispatched to the Windows Service Control Manager. The same binary + // serves both, detected at runtime rather than selected by a flag. + err := service.Run(func(ctx context.Context) error { + return application.Run(ctx) + }) + + if err != nil { + fmt.Fprintf(os.Stderr, "docksight-agent: %v\n", err) + os.Exit(1) + } +} diff --git a/apps/agent/internal/app/app.go b/apps/agent/internal/app/app.go index 543c974..a7709aa 100644 --- a/apps/agent/internal/app/app.go +++ b/apps/agent/internal/app/app.go @@ -1,153 +1,153 @@ -package app - -import ( - "context" - "fmt" - "os" - "runtime" - - "docksight-agent/internal/communication" - "docksight-agent/internal/config" - "docksight-agent/internal/docker" - "docksight-agent/internal/identity" - "docksight-agent/internal/lifecycle" - "docksight-agent/internal/logger" - "docksight-agent/internal/logs" - "docksight-agent/internal/version" -) - -// App is the DockSight agent application bootstrapper. -type App struct { - configPath string -} - -// New creates an application instance. -func New(configPath string) *App { - if configPath == "" { - configPath = "config.yaml" - } - return &App{configPath: configPath} -} - -// Run executes the agent lifecycle: -// config → logger → identity → docker → logs → connect → register → heartbeat → wait. -// -// The context is the supervisor's: cancelling it shuts the agent down exactly -// as an interrupt does. Under the Windows Service Control Manager that is the -// only way in, because a service never receives a signal. -func (a *App) Run(ctx context.Context) error { - cfg, err := config.Load(a.configPath) - if err != nil { - return fmt.Errorf("configuration: %w", err) - } - - sink, closeSink, err := openLogSink() - if err != nil { - return fmt.Errorf("logging: %w", err) - } - if closeSink != nil { - defer closeSink() - } - - logger.Setup(cfg.Logging.Level, sink) - logger.Info("configuration loaded", "path", a.configPath, "server", cfg.Server.URL) - warnIfPlaintextServerURL(cfg.Server.URL) - logger.Printf("DockSight Agent started\n") - - id, created, err := identity.LoadOrCreate(cfg.Agent.IdentityFile) - if err != nil { - return fmt.Errorf("identity: %w", err) - } - if created { - logger.Info("identity created", "id", id.ID, "path", cfg.Agent.IdentityFile) - } else { - logger.Info("identity loaded", "id", id.ID, "path", cfg.Agent.IdentityFile) - } - - socket := cfg.Docker.Socket - if socket == "" { - socket = docker.DefaultSocket() - } - - dockerAvailable := docker.SocketExists(socket) - var dockerService *docker.Service - var dockerClient *docker.Client - - if dockerAvailable { - logger.Info("docker socket found", "socket", socket) - dockerClient, err = docker.NewClient(socket) - if err != nil { - logger.Warn("docker client init failed", "error", err.Error()) - } else { - dockerService = docker.NewService(dockerClient) - pingCtx := context.Background() - if pingErr := dockerService.Ping(pingCtx); pingErr != nil { - logger.Warn("docker engine not reachable", "error", pingErr.Error()) - } else if info, infoErr := dockerService.GetDockerInfo(pingCtx); infoErr == nil { - logger.Info("docker engine available", - "version", info.Version, - "os", info.OS, - "arch", info.Architecture, - ) - } - } - } else { - logger.Warn("docker socket not found", "socket", socket) - } - - var logsService *logs.Service - if dockerService != nil { - logsService = logs.NewService(dockerService) - } - - hostname, _ := os.Hostname() - if hostname == "" { - hostname = "unknown" - } - - printStartupSummary(id.ID, created, dockerAvailable, cfg.Server.URL) - - lc := lifecycle.New(ctx) - client := communication.NewClient(cfg.Server.URL, communication.AgentInfo{ - UUID: id.ID, - Hostname: hostname, - OS: runtime.GOOS, - Architecture: runtime.GOARCH, - Version: version.Version, - }, dockerService, logsService) - - go client.Run(lc.Context()) - - logger.Info("agent ready; waiting for shutdown signal", - "version", version.Version, - "agentId", id.ID, - ) - lc.Wait(func() { - if logsService != nil { - logsService.Close() - } - if dockerClient != nil { - _ = dockerClient.Close() - } - logger.Info("agent stopped") - }) - - return nil -} - -func printStartupSummary(agentID string, identityCreated bool, dockerAvailable bool, serverURL string) { - logger.Printf("\nDockSight Agent %s\n\n", version.String()) - logger.Printf("Configuration loaded\n") - if identityCreated { - logger.Printf("Identity created (%s)\n", agentID) - } else { - logger.Printf("Identity loaded (%s)\n", agentID) - } - if dockerAvailable { - logger.Printf("Docker socket found\n") - } else { - logger.Printf("Docker socket not found\n") - } - logger.Printf("Server: %s\n", serverURL) - logger.Printf("\nAgent Status: READY\n\n") -} +package app + +import ( + "context" + "fmt" + "os" + "runtime" + + "docksight-agent/internal/communication" + "docksight-agent/internal/config" + "docksight-agent/internal/docker" + "docksight-agent/internal/identity" + "docksight-agent/internal/lifecycle" + "docksight-agent/internal/logger" + "docksight-agent/internal/logs" + "docksight-agent/internal/version" +) + +// App is the DockSight agent application bootstrapper. +type App struct { + configPath string +} + +// New creates an application instance. +func New(configPath string) *App { + if configPath == "" { + configPath = "config.yaml" + } + return &App{configPath: configPath} +} + +// Run executes the agent lifecycle: +// config → logger → identity → docker → logs → connect → register → heartbeat → wait. +// +// The context is the supervisor's: cancelling it shuts the agent down exactly +// as an interrupt does. Under the Windows Service Control Manager that is the +// only way in, because a service never receives a signal. +func (a *App) Run(ctx context.Context) error { + cfg, err := config.Load(a.configPath) + if err != nil { + return fmt.Errorf("configuration: %w", err) + } + + sink, closeSink, err := openLogSink() + if err != nil { + return fmt.Errorf("logging: %w", err) + } + if closeSink != nil { + defer closeSink() + } + + logger.Setup(cfg.Logging.Level, sink) + logger.Info("configuration loaded", "path", a.configPath, "server", cfg.Server.URL) + warnIfPlaintextServerURL(cfg.Server.URL) + logger.Printf("DockSight Agent started\n") + + id, created, err := identity.LoadOrCreate(cfg.Agent.IdentityFile) + if err != nil { + return fmt.Errorf("identity: %w", err) + } + if created { + logger.Info("identity created", "id", id.ID, "path", cfg.Agent.IdentityFile) + } else { + logger.Info("identity loaded", "id", id.ID, "path", cfg.Agent.IdentityFile) + } + + socket := cfg.Docker.Socket + if socket == "" { + socket = docker.DefaultSocket() + } + + dockerAvailable := docker.SocketExists(socket) + var dockerService *docker.Service + var dockerClient *docker.Client + + if dockerAvailable { + logger.Info("docker socket found", "socket", socket) + dockerClient, err = docker.NewClient(socket) + if err != nil { + logger.Warn("docker client init failed", "error", err.Error()) + } else { + dockerService = docker.NewService(dockerClient) + pingCtx := context.Background() + if pingErr := dockerService.Ping(pingCtx); pingErr != nil { + logger.Warn("docker engine not reachable", "error", pingErr.Error()) + } else if info, infoErr := dockerService.GetDockerInfo(pingCtx); infoErr == nil { + logger.Info("docker engine available", + "version", info.Version, + "os", info.OS, + "arch", info.Architecture, + ) + } + } + } else { + logger.Warn("docker socket not found", "socket", socket) + } + + var logsService *logs.Service + if dockerService != nil { + logsService = logs.NewService(dockerService) + } + + hostname, _ := os.Hostname() + if hostname == "" { + hostname = "unknown" + } + + printStartupSummary(id.ID, created, dockerAvailable, cfg.Server.URL) + + lc := lifecycle.New(ctx) + client := communication.NewClient(cfg.Server.URL, communication.AgentInfo{ + UUID: id.ID, + Hostname: hostname, + OS: runtime.GOOS, + Architecture: runtime.GOARCH, + Version: version.Version, + }, dockerService, logsService) + + go client.Run(lc.Context()) + + logger.Info("agent ready; waiting for shutdown signal", + "version", version.Version, + "agentId", id.ID, + ) + lc.Wait(func() { + if logsService != nil { + logsService.Close() + } + if dockerClient != nil { + _ = dockerClient.Close() + } + logger.Info("agent stopped") + }) + + return nil +} + +func printStartupSummary(agentID string, identityCreated bool, dockerAvailable bool, serverURL string) { + logger.Printf("\nDockSight Agent %s\n\n", version.String()) + logger.Printf("Configuration loaded\n") + if identityCreated { + logger.Printf("Identity created (%s)\n", agentID) + } else { + logger.Printf("Identity loaded (%s)\n", agentID) + } + if dockerAvailable { + logger.Printf("Docker socket found\n") + } else { + logger.Printf("Docker socket not found\n") + } + logger.Printf("Server: %s\n", serverURL) + logger.Printf("\nAgent Status: READY\n\n") +} diff --git a/apps/agent/internal/communication/client.go b/apps/agent/internal/communication/client.go index 0c7e8aa..1d0eec9 100644 --- a/apps/agent/internal/communication/client.go +++ b/apps/agent/internal/communication/client.go @@ -73,13 +73,13 @@ type HostMetricsPayload struct { // ContainerSummary matches protocol container discovery fields. type ContainerSummary struct { - ID string `json:"id"` - Name string `json:"name"` - Image string `json:"image"` - Status string `json:"status"` - State string `json:"state"` - Ports []container.Port `json:"ports"` - Created int64 `json:"created"` + ID string `json:"id"` + Name string `json:"name"` + Image string `json:"image"` + Status string `json:"status"` + State string `json:"state"` + Ports []container.Port `json:"ports"` + Created int64 `json:"created"` } // ContainerListedPayload is sent on container.listed. @@ -469,14 +469,13 @@ func (c *Client) handleContainerList(ctx context.Context, conn *websocket.Conn) summaries := make([]ContainerSummary, 0, len(items)) for _, item := range items { summaries = append(summaries, ContainerSummary{ - ID: item.ID, - Name: item.Name, - Image: item.Image, - Status: item.Status, - State: item.State, - Ports: item.Ports, - Created : item.Created, - + ID: item.ID, + Name: item.Name, + Image: item.Image, + Status: item.Status, + State: item.State, + Ports: item.Ports, + Created: item.Created, }) } diff --git a/apps/agent/internal/config/config.go b/apps/agent/internal/config/config.go index 6eeff08..773b142 100644 --- a/apps/agent/internal/config/config.go +++ b/apps/agent/internal/config/config.go @@ -1,118 +1,118 @@ -package config - -import ( - "fmt" - "os" - "path/filepath" - "runtime" - - "gopkg.in/yaml.v3" -) - -// Config is the agent runtime configuration loaded from config.yaml. -type Config struct { - Agent AgentConfig `yaml:"agent"` - Server ServerConfig `yaml:"server"` - Docker DockerConfig `yaml:"docker"` - Logging LoggingConfig `yaml:"logging"` -} - -// AgentConfig controls identity persistence and working directories. -type AgentConfig struct { - DataDir string `yaml:"data_dir"` - IdentityFile string `yaml:"identity_file"` -} - -// ServerConfig controls DockSight server connectivity. -type ServerConfig struct { - URL string `yaml:"url"` -} - -// DockerConfig holds local Docker Engine access settings. -type DockerConfig struct { - Socket string `yaml:"socket"` -} - -// LoggingConfig controls log verbosity. -type LoggingConfig struct { - Level string `yaml:"level"` -} - -// Load reads configuration from the given YAML path (default: config.yaml). -func Load(path string) (*Config, error) { - if path == "" { - path = "config.yaml" - } - - data, err := os.ReadFile(path) - if err != nil { - return nil, fmt.Errorf("read config %q: %w", path, err) - } - - cfg := defaultConfig() - if err := yaml.Unmarshal(data, cfg); err != nil { - return nil, fmt.Errorf("parse config %q: %w", path, err) - } - - cfg.applyDefaults() - if err := cfg.validate(); err != nil { - return nil, fmt.Errorf("validate config %q: %w", path, err) - } - - return cfg, nil -} - -func defaultConfig() *Config { - return &Config{ - Agent: AgentConfig{ - DataDir: "./data", - IdentityFile: "./data/identity.json", - }, - Docker: DockerConfig{ - Socket: defaultDockerSocket(), - }, - Logging: LoggingConfig{ - Level: "info", - }, - } -} - -func (c *Config) applyDefaults() { - defaults := defaultConfig() - - if c.Agent.DataDir == "" { - c.Agent.DataDir = defaults.Agent.DataDir - } - if c.Agent.IdentityFile == "" { - c.Agent.IdentityFile = filepath.Join(c.Agent.DataDir, "identity.json") - } - if c.Server.URL == "" { - c.Server.URL = os.Getenv("AGENT_SERVER_URL") - } - if c.Docker.Socket == "" { - c.Docker.Socket = defaults.Docker.Socket - } - if c.Logging.Level == "" { - c.Logging.Level = defaults.Logging.Level - } -} - -func (c *Config) validate() error { - if c.Agent.IdentityFile == "" { - return fmt.Errorf("agent.identity_file is required") - } - if c.Server.URL == "" { - return fmt.Errorf("server.url is required when AGENT_SERVER_URL is not set") - } - if c.Docker.Socket == "" { - return fmt.Errorf("docker.socket is required") - } - return nil -} - -func defaultDockerSocket() string { - if runtime.GOOS == "windows" { - return `\\.\pipe\docker_engine` - } - return "/var/run/docker.sock" -} +package config + +import ( + "fmt" + "os" + "path/filepath" + "runtime" + + "gopkg.in/yaml.v3" +) + +// Config is the agent runtime configuration loaded from config.yaml. +type Config struct { + Agent AgentConfig `yaml:"agent"` + Server ServerConfig `yaml:"server"` + Docker DockerConfig `yaml:"docker"` + Logging LoggingConfig `yaml:"logging"` +} + +// AgentConfig controls identity persistence and working directories. +type AgentConfig struct { + DataDir string `yaml:"data_dir"` + IdentityFile string `yaml:"identity_file"` +} + +// ServerConfig controls DockSight server connectivity. +type ServerConfig struct { + URL string `yaml:"url"` +} + +// DockerConfig holds local Docker Engine access settings. +type DockerConfig struct { + Socket string `yaml:"socket"` +} + +// LoggingConfig controls log verbosity. +type LoggingConfig struct { + Level string `yaml:"level"` +} + +// Load reads configuration from the given YAML path (default: config.yaml). +func Load(path string) (*Config, error) { + if path == "" { + path = "config.yaml" + } + + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read config %q: %w", path, err) + } + + cfg := defaultConfig() + if err := yaml.Unmarshal(data, cfg); err != nil { + return nil, fmt.Errorf("parse config %q: %w", path, err) + } + + cfg.applyDefaults() + if err := cfg.validate(); err != nil { + return nil, fmt.Errorf("validate config %q: %w", path, err) + } + + return cfg, nil +} + +func defaultConfig() *Config { + return &Config{ + Agent: AgentConfig{ + DataDir: "./data", + IdentityFile: "./data/identity.json", + }, + Docker: DockerConfig{ + Socket: defaultDockerSocket(), + }, + Logging: LoggingConfig{ + Level: "info", + }, + } +} + +func (c *Config) applyDefaults() { + defaults := defaultConfig() + + if c.Agent.DataDir == "" { + c.Agent.DataDir = defaults.Agent.DataDir + } + if c.Agent.IdentityFile == "" { + c.Agent.IdentityFile = filepath.Join(c.Agent.DataDir, "identity.json") + } + if c.Server.URL == "" { + c.Server.URL = os.Getenv("AGENT_SERVER_URL") + } + if c.Docker.Socket == "" { + c.Docker.Socket = defaults.Docker.Socket + } + if c.Logging.Level == "" { + c.Logging.Level = defaults.Logging.Level + } +} + +func (c *Config) validate() error { + if c.Agent.IdentityFile == "" { + return fmt.Errorf("agent.identity_file is required") + } + if c.Server.URL == "" { + return fmt.Errorf("server.url is required when AGENT_SERVER_URL is not set") + } + if c.Docker.Socket == "" { + return fmt.Errorf("docker.socket is required") + } + return nil +} + +func defaultDockerSocket() string { + if runtime.GOOS == "windows" { + return `\\.\pipe\docker_engine` + } + return "/var/run/docker.sock" +} diff --git a/apps/agent/internal/docker/client.go b/apps/agent/internal/docker/client.go index 4b7a7d2..ca43182 100644 --- a/apps/agent/internal/docker/client.go +++ b/apps/agent/internal/docker/client.go @@ -1,91 +1,91 @@ -package docker - -import ( - "context" - "fmt" - "net" - "os" - "runtime" - - "github.com/docker/docker/client" -) - -// Client wraps the Docker Engine Go SDK with a platform-aware dialer. -type Client struct { - sdk *client.Client - socket string -} - -// NewClient connects to Docker Engine using the configured socket/named pipe. -func NewClient(socket string) (*Client, error) { - if socket == "" { - socket = DefaultSocket() - } - - opts := []client.Opt{ - client.WithHost(engineHost(socket)), - client.WithAPIVersionNegotiation(), - // Apply after WithHost so we override any transport dialer ConfigureTransport set. - client.WithDialContext(func(ctx context.Context, _, _ string) (net.Conn, error) { - return dialDocker(ctx, socket) - }), - } - - sdk, err := client.NewClientWithOpts(opts...) - if err != nil { - return nil, fmt.Errorf("docker sdk client: %w", err) - } - - return &Client{ - sdk: sdk, - socket: socket, - }, nil -} - -// SDK returns the underlying Docker Engine API client. -func (c *Client) SDK() *client.Client { - return c.sdk -} - -// DefaultSocket returns the platform Docker Engine endpoint path. -func DefaultSocket() string { - if runtime.GOOS == "windows" { - return `\\.\pipe\docker_engine` - } - return "/var/run/docker.sock" -} - -func engineHost(socket string) string { - if runtime.GOOS == "windows" { - if socket == "" || socket == `\\.\pipe\docker_engine` { - return client.DefaultDockerHost - } - return "npipe://" + socket - } - if socket == "" { - socket = "/var/run/docker.sock" - } - return "unix://" + socket -} - -// Close releases SDK resources. -func (c *Client) Close() error { - if c.sdk == nil { - return nil - } - return c.sdk.Close() -} - -// Socket returns the configured engine socket path. -func (c *Client) Socket() string { - return c.socket -} - -// SocketExists reports whether the configured Docker socket/pipe path exists. -func SocketExists(socket string) bool { - if socket == "" { - socket = DefaultSocket() - } - _, err := os.Stat(socket) - return err == nil -} +package docker + +import ( + "context" + "fmt" + "net" + "os" + "runtime" + + "github.com/docker/docker/client" +) + +// Client wraps the Docker Engine Go SDK with a platform-aware dialer. +type Client struct { + sdk *client.Client + socket string +} + +// NewClient connects to Docker Engine using the configured socket/named pipe. +func NewClient(socket string) (*Client, error) { + if socket == "" { + socket = DefaultSocket() + } + + opts := []client.Opt{ + client.WithHost(engineHost(socket)), + client.WithAPIVersionNegotiation(), + // Apply after WithHost so we override any transport dialer ConfigureTransport set. + client.WithDialContext(func(ctx context.Context, _, _ string) (net.Conn, error) { + return dialDocker(ctx, socket) + }), + } + + sdk, err := client.NewClientWithOpts(opts...) + if err != nil { + return nil, fmt.Errorf("docker sdk client: %w", err) + } + + return &Client{ + sdk: sdk, + socket: socket, + }, nil +} + +// SDK returns the underlying Docker Engine API client. +func (c *Client) SDK() *client.Client { + return c.sdk +} + +// DefaultSocket returns the platform Docker Engine endpoint path. +func DefaultSocket() string { + if runtime.GOOS == "windows" { + return `\\.\pipe\docker_engine` + } + return "/var/run/docker.sock" +} + +func engineHost(socket string) string { + if runtime.GOOS == "windows" { + if socket == "" || socket == `\\.\pipe\docker_engine` { + return client.DefaultDockerHost + } + return "npipe://" + socket + } + if socket == "" { + socket = "/var/run/docker.sock" + } + return "unix://" + socket +} + +// Close releases SDK resources. +func (c *Client) Close() error { + if c.sdk == nil { + return nil + } + return c.sdk.Close() +} + +// Socket returns the configured engine socket path. +func (c *Client) Socket() string { + return c.socket +} + +// SocketExists reports whether the configured Docker socket/pipe path exists. +func SocketExists(socket string) bool { + if socket == "" { + socket = DefaultSocket() + } + _, err := os.Stat(socket) + return err == nil +} diff --git a/apps/agent/internal/docker/service.go b/apps/agent/internal/docker/service.go index 0688e42..85c671c 100644 --- a/apps/agent/internal/docker/service.go +++ b/apps/agent/internal/docker/service.go @@ -67,12 +67,12 @@ func (s *Service) ListContainers(ctx context.Context) ([]Container, error) { name = strings.TrimPrefix(item.Names[0], "/") } result = append(result, Container{ - ID: item.ID, - Name: name, - Image: item.Image, - Status: item.Status, - State: item.State, - Ports: item.Ports, + ID: item.ID, + Name: name, + Image: item.Image, + Status: item.Status, + State: item.State, + Ports: item.Ports, Created: item.Created, }) } @@ -248,7 +248,6 @@ func mapContainerInspect(container types.ContainerJSON) *ContainerInspect { RestartPolicy: string(container.HostConfig.RestartPolicy.Name), Entrypoint: container.Config.Entrypoint, Env: container.Config.Env, - } } diff --git a/apps/agent/internal/docker/types.go b/apps/agent/internal/docker/types.go index 7c58702..4c0ed03 100644 --- a/apps/agent/internal/docker/types.go +++ b/apps/agent/internal/docker/types.go @@ -15,13 +15,13 @@ type Info struct { // Container is a read-only discovery summary. type Container struct { - ID string `json:"id"` - Name string `json:"name"` - Image string `json:"image"` - Status string `json:"status"` - State string `json:"state"` - Ports []container.Port `json:"ports"` - Created int64 `json:"createdAt"` + ID string `json:"id"` + Name string `json:"name"` + Image string `json:"image"` + Status string `json:"status"` + State string `json:"state"` + Ports []container.Port `json:"ports"` + Created int64 `json:"createdAt"` } type Port struct { Private int `json:"private"` @@ -64,5 +64,4 @@ type ContainerInspect struct { RestartPolicy string `json:"restartPolicy"` Entrypoint []string `json:"entrypoint"` Env []string `json:"env"` - } diff --git a/apps/agent/internal/identity/identity.go b/apps/agent/internal/identity/identity.go index 7f99e6d..c21e58a 100644 --- a/apps/agent/internal/identity/identity.go +++ b/apps/agent/internal/identity/identity.go @@ -1,97 +1,97 @@ -package identity - -import ( - "crypto/rand" - "encoding/json" - "fmt" - "os" - "path/filepath" - "time" -) - -// Identity is the durable agent identity persisted on disk. -type Identity struct { - ID string `json:"id"` - CreatedAt time.Time `json:"created_at"` -} - -// LoadOrCreate returns an existing identity from path, or generates and persists a new one. -func LoadOrCreate(path string) (*Identity, bool, error) { - if path == "" { - return nil, false, fmt.Errorf("identity path is required") - } - - if _, err := os.Stat(path); err == nil { - id, err := load(path) - if err != nil { - return nil, false, err - } - return id, false, nil - } else if !os.IsNotExist(err) { - return nil, false, fmt.Errorf("stat identity file: %w", err) - } - - id, err := newIdentity() - if err != nil { - return nil, false, err - } - if err := save(path, id); err != nil { - return nil, false, err - } - return id, true, nil -} - -func load(path string) (*Identity, error) { - data, err := os.ReadFile(path) - if err != nil { - return nil, fmt.Errorf("read identity file: %w", err) - } - - var id Identity - if err := json.Unmarshal(data, &id); err != nil { - return nil, fmt.Errorf("parse identity file: %w", err) - } - if id.ID == "" { - return nil, fmt.Errorf("identity file %q is missing id", path) - } - return &id, nil -} - -func save(path string, id *Identity) error { - if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { - return fmt.Errorf("create identity directory: %w", err) - } - - data, err := json.MarshalIndent(id, "", " ") - if err != nil { - return fmt.Errorf("encode identity: %w", err) - } - data = append(data, '\n') - - if err := os.WriteFile(path, data, 0o600); err != nil { - return fmt.Errorf("write identity file: %w", err) - } - return nil -} - -func newIdentity() (*Identity, error) { - uuid, err := newUUID() - if err != nil { - return nil, err - } - return &Identity{ - ID: uuid, - CreatedAt: time.Now().UTC(), - }, nil -} - -// newUUID generates an RFC 4122 version 4 UUID. -func newUUID() (string, error) { - var b [16]byte - if _, err := rand.Read(b[:]); err != nil { - return "", fmt.Errorf("generate uuid: %w", err) - } - b[6] = (b[6] & 0x0f) | 0x40 - b[8] = (b[8] & 0x3f) | 0x80 - return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:]), nil -} +package identity + +import ( + "crypto/rand" + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" +) + +// Identity is the durable agent identity persisted on disk. +type Identity struct { + ID string `json:"id"` + CreatedAt time.Time `json:"created_at"` +} + +// LoadOrCreate returns an existing identity from path, or generates and persists a new one. +func LoadOrCreate(path string) (*Identity, bool, error) { + if path == "" { + return nil, false, fmt.Errorf("identity path is required") + } + + if _, err := os.Stat(path); err == nil { + id, err := load(path) + if err != nil { + return nil, false, err + } + return id, false, nil + } else if !os.IsNotExist(err) { + return nil, false, fmt.Errorf("stat identity file: %w", err) + } + + id, err := newIdentity() + if err != nil { + return nil, false, err + } + if err := save(path, id); err != nil { + return nil, false, err + } + return id, true, nil +} + +func load(path string) (*Identity, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read identity file: %w", err) + } + + var id Identity + if err := json.Unmarshal(data, &id); err != nil { + return nil, fmt.Errorf("parse identity file: %w", err) + } + if id.ID == "" { + return nil, fmt.Errorf("identity file %q is missing id", path) + } + return &id, nil +} + +func save(path string, id *Identity) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("create identity directory: %w", err) + } + + data, err := json.MarshalIndent(id, "", " ") + if err != nil { + return fmt.Errorf("encode identity: %w", err) + } + data = append(data, '\n') + + if err := os.WriteFile(path, data, 0o600); err != nil { + return fmt.Errorf("write identity file: %w", err) + } + return nil +} + +func newIdentity() (*Identity, error) { + uuid, err := newUUID() + if err != nil { + return nil, err + } + return &Identity{ + ID: uuid, + CreatedAt: time.Now().UTC(), + }, nil +} + +// newUUID generates an RFC 4122 version 4 UUID. +func newUUID() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", fmt.Errorf("generate uuid: %w", err) + } + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:]), nil +} diff --git a/apps/agent/internal/lifecycle/lifecycle.go b/apps/agent/internal/lifecycle/lifecycle.go index 8267a99..c9b7229 100644 --- a/apps/agent/internal/lifecycle/lifecycle.go +++ b/apps/agent/internal/lifecycle/lifecycle.go @@ -1,58 +1,58 @@ -package lifecycle - -import ( - "context" - "os" - "os/signal" - "sync" - "syscall" - - "docksight-agent/internal/logger" -) - -// Manager coordinates graceful startup/shutdown for the agent process. -type Manager struct { - ctx context.Context - cancel context.CancelFunc - once sync.Once -} - -// New creates a lifecycle manager that listens for SIGINT and SIGTERM, and -// also shuts down when parent is cancelled. -// -// The parent is what makes the same shutdown path reachable from outside the -// process. A Windows service never receives SIGTERM — the Service Control -// Manager delivers SERVICE_CONTROL_STOP on its own channel — so the service -// handler cancels the parent and the hooks below run exactly as they do for -// Ctrl+C. One shutdown path, two triggers. -func New(parent context.Context) *Manager { - if parent == nil { - parent = context.Background() - } - ctx, cancel := signal.NotifyContext(parent, os.Interrupt, syscall.SIGTERM) - return &Manager{ctx: ctx, cancel: cancel} -} - -// Context returns the lifecycle context cancelled on interrupt. -func (m *Manager) Context() context.Context { - return m.ctx -} - -// Wait blocks until SIGINT/SIGTERM is received, then runs optional shutdown hooks. -func (m *Manager) Wait(onShutdown ...func()) { - <-m.ctx.Done() - logger.Info("shutdown signal received") - m.Shutdown(onShutdown...) -} - -// Shutdown runs shutdown hooks exactly once and cancels the lifecycle context. -func (m *Manager) Shutdown(hooks ...func()) { - m.once.Do(func() { - for _, hook := range hooks { - if hook != nil { - hook() - } - } - m.cancel() - }) -} +package lifecycle + +import ( + "context" + "os" + "os/signal" + "sync" + "syscall" + + "docksight-agent/internal/logger" +) + +// Manager coordinates graceful startup/shutdown for the agent process. +type Manager struct { + ctx context.Context + cancel context.CancelFunc + once sync.Once +} + +// New creates a lifecycle manager that listens for SIGINT and SIGTERM, and +// also shuts down when parent is cancelled. +// +// The parent is what makes the same shutdown path reachable from outside the +// process. A Windows service never receives SIGTERM — the Service Control +// Manager delivers SERVICE_CONTROL_STOP on its own channel — so the service +// handler cancels the parent and the hooks below run exactly as they do for +// Ctrl+C. One shutdown path, two triggers. +func New(parent context.Context) *Manager { + if parent == nil { + parent = context.Background() + } + ctx, cancel := signal.NotifyContext(parent, os.Interrupt, syscall.SIGTERM) + return &Manager{ctx: ctx, cancel: cancel} +} + +// Context returns the lifecycle context cancelled on interrupt. +func (m *Manager) Context() context.Context { + return m.ctx +} + +// Wait blocks until SIGINT/SIGTERM is received, then runs optional shutdown hooks. +func (m *Manager) Wait(onShutdown ...func()) { + <-m.ctx.Done() + logger.Info("shutdown signal received") + m.Shutdown(onShutdown...) +} + +// Shutdown runs shutdown hooks exactly once and cancels the lifecycle context. +func (m *Manager) Shutdown(hooks ...func()) { + m.once.Do(func() { + for _, hook := range hooks { + if hook != nil { + hook() + } + } + m.cancel() + }) +} diff --git a/apps/agent/internal/logger/logger.go b/apps/agent/internal/logger/logger.go index 77a454e..f47dd1a 100644 --- a/apps/agent/internal/logger/logger.go +++ b/apps/agent/internal/logger/logger.go @@ -1,98 +1,98 @@ -package logger - -import ( - "fmt" - "io" - "log/slog" - "os" - "strings" - "sync" - "time" -) - -var ( - defaultLogger *slog.Logger = slog.Default() - - // output is where Printf writes. Tracked separately from the slog handler - // because Printf predates Setup taking a writer and used fmt.Printf, which - // goes to stdout unconditionally — under a Windows service there is no - // console, so the entire startup summary was written to a handle nobody - // could read. - output io.Writer = os.Stdout - - // outputMu guards output. Setup runs during startup while nothing else is - // logging, but Printf is reachable from any goroutine afterwards. - outputMu sync.RWMutex -) - -// Setup configures the package-level structured logger. -func Setup(level string, out io.Writer) { - if out == nil { - out = os.Stdout - } - - outputMu.Lock() - output = out - outputMu.Unlock() - - opts := &slog.HandlerOptions{ - Level: parseLevel(level), - ReplaceAttr: func(_ []string, attr slog.Attr) slog.Attr { - if attr.Key == slog.TimeKey { - return slog.String(slog.TimeKey, attr.Value.Time().UTC().Format(time.RFC3339)) - } - return attr - }, - } - - defaultLogger = slog.New(slog.NewTextHandler(out, opts)) -} - -func parseLevel(level string) slog.Level { - switch strings.ToLower(strings.TrimSpace(level)) { - case "debug": - return slog.LevelDebug - case "warn", "warning": - return slog.LevelWarn - case "error": - return slog.LevelError - default: - return slog.LevelInfo - } -} - -// Debug logs a debug-level message. -func Debug(msg string, args ...any) { - defaultLogger.Debug(msg, args...) -} - -// Info logs an info-level message. -func Info(msg string, args ...any) { - defaultLogger.Info(msg, args...) -} - -// Warn logs a warning-level message. -func Warn(msg string, args ...any) { - defaultLogger.Warn(msg, args...) -} - -// Error logs an error-level message. -func Error(msg string, args ...any) { - defaultLogger.Error(msg, args...) -} - -// Fatal logs an error-level message and exits the process. -func Fatal(msg string, args ...any) { - defaultLogger.Error(msg, args...) - os.Exit(1) -} - -// Printf writes a plain line to the configured output (used for the human -// startup summary). -func Printf(format string, args ...any) { - outputMu.RLock() - out := output - outputMu.RUnlock() - - fmt.Fprintf(out, format, args...) -} +package logger + +import ( + "fmt" + "io" + "log/slog" + "os" + "strings" + "sync" + "time" +) + +var ( + defaultLogger *slog.Logger = slog.Default() + + // output is where Printf writes. Tracked separately from the slog handler + // because Printf predates Setup taking a writer and used fmt.Printf, which + // goes to stdout unconditionally — under a Windows service there is no + // console, so the entire startup summary was written to a handle nobody + // could read. + output io.Writer = os.Stdout + + // outputMu guards output. Setup runs during startup while nothing else is + // logging, but Printf is reachable from any goroutine afterwards. + outputMu sync.RWMutex +) + +// Setup configures the package-level structured logger. +func Setup(level string, out io.Writer) { + if out == nil { + out = os.Stdout + } + + outputMu.Lock() + output = out + outputMu.Unlock() + + opts := &slog.HandlerOptions{ + Level: parseLevel(level), + ReplaceAttr: func(_ []string, attr slog.Attr) slog.Attr { + if attr.Key == slog.TimeKey { + return slog.String(slog.TimeKey, attr.Value.Time().UTC().Format(time.RFC3339)) + } + return attr + }, + } + + defaultLogger = slog.New(slog.NewTextHandler(out, opts)) +} + +func parseLevel(level string) slog.Level { + switch strings.ToLower(strings.TrimSpace(level)) { + case "debug": + return slog.LevelDebug + case "warn", "warning": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return slog.LevelInfo + } +} + +// Debug logs a debug-level message. +func Debug(msg string, args ...any) { + defaultLogger.Debug(msg, args...) +} + +// Info logs an info-level message. +func Info(msg string, args ...any) { + defaultLogger.Info(msg, args...) +} + +// Warn logs a warning-level message. +func Warn(msg string, args ...any) { + defaultLogger.Warn(msg, args...) +} + +// Error logs an error-level message. +func Error(msg string, args ...any) { + defaultLogger.Error(msg, args...) +} + +// Fatal logs an error-level message and exits the process. +func Fatal(msg string, args ...any) { + defaultLogger.Error(msg, args...) + os.Exit(1) +} + +// Printf writes a plain line to the configured output (used for the human +// startup summary). +func Printf(format string, args ...any) { + outputMu.RLock() + out := output + outputMu.RUnlock() + + fmt.Fprintf(out, format, args...) +} diff --git a/apps/agent/internal/logs/decode.go b/apps/agent/internal/logs/decode.go index c6370b2..9374f69 100644 --- a/apps/agent/internal/logs/decode.go +++ b/apps/agent/internal/logs/decode.go @@ -1,149 +1,149 @@ -package logs - -import ( - "bufio" - "encoding/binary" - "fmt" - "io" - "strings" - "time" -) - -const ( - streamStdout = 1 - streamStderr = 2 -) - -// DecodeLogStream reads a Docker Engine log stream and invokes onEntry for each line. -// Supports multiplexed (non-TTY) frames and plain line-based (TTY) streams. -func DecodeLogStream(r io.Reader, onEntry func(Entry) error) error { - br := bufio.NewReader(r) - header, err := br.Peek(8) - if err != nil { - if err == io.EOF { - return nil - } - return fmt.Errorf("peek log header: %w", err) - } - - // Multiplexed streams start with stream type 0/1/2 and a big-endian size. - if isLikelyMultiplexed(header) { - return decodeMultiplexed(br, onEntry) - } - return decodePlainLines(br, "stdout", onEntry) -} - -func isLikelyMultiplexed(header []byte) bool { - if len(header) < 8 { - return false - } - streamType := header[0] - if streamType > 2 { - return false - } - // Bytes 1-3 are reserved zeroes in the Docker multiplex header. - if header[1] != 0 || header[2] != 0 || header[3] != 0 { - return false - } - return true -} - -func decodeMultiplexed(br *bufio.Reader, onEntry func(Entry) error) error { - header := make([]byte, 8) - for { - if _, err := io.ReadFull(br, header); err != nil { - if err == io.EOF || err == io.ErrUnexpectedEOF { - return nil - } - return fmt.Errorf("read multiplex header: %w", err) - } - - streamName := streamName(header[0]) - size := binary.BigEndian.Uint32(header[4:8]) - if size == 0 { - continue - } - - payload := make([]byte, size) - if _, err := io.ReadFull(br, payload); err != nil { - if err == io.EOF || err == io.ErrUnexpectedEOF { - return nil - } - return fmt.Errorf("read multiplex payload: %w", err) - } - - if err := emitLines(string(payload), streamName, onEntry); err != nil { - return err - } - } -} - -func decodePlainLines(br *bufio.Reader, stream string, onEntry func(Entry) error) error { - for { - line, err := br.ReadString('\n') - if len(line) > 0 { - if err := emitLines(line, stream, onEntry); err != nil { - return err - } - } - if err != nil { - if err == io.EOF { - return nil - } - return fmt.Errorf("read plain log line: %w", err) - } - } -} - -func emitLines(payload string, stream string, onEntry func(Entry) error) error { - payload = strings.ReplaceAll(payload, "\r\n", "\n") - parts := strings.Split(payload, "\n") - for i, part := range parts { - // Keep empty trailing segment only when payload did not end with newline? - // Drop empty fragments from trailing newlines. - if part == "" && i == len(parts)-1 { - continue - } - entry := ParseTimestampedLine(part, stream) - if err := onEntry(entry); err != nil { - return err - } - } - return nil -} - -// ParseTimestampedLine splits a Docker `--timestamps` log line into timestamp + message. -func ParseTimestampedLine(line string, stream string) Entry { - line = strings.TrimRight(line, "\r\n") - timestamp := time.Now().UTC().Format(time.RFC3339Nano) - message := line - - space := strings.IndexByte(line, ' ') - if space > 0 { - candidate := line[:space] - if _, err := time.Parse(time.RFC3339Nano, candidate); err == nil { - timestamp = candidate - message = line[space+1:] - } else if t, err := time.Parse(time.RFC3339, candidate); err == nil { - timestamp = t.UTC().Format(time.RFC3339Nano) - message = line[space+1:] - } - } - - return Entry{ - Timestamp: timestamp, - Stream: stream, - Message: message, - } -} - -func streamName(code byte) string { - switch code { - case streamStdout: - return "stdout" - case streamStderr: - return "stderr" - default: - return "stdout" - } -} +package logs + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" + "strings" + "time" +) + +const ( + streamStdout = 1 + streamStderr = 2 +) + +// DecodeLogStream reads a Docker Engine log stream and invokes onEntry for each line. +// Supports multiplexed (non-TTY) frames and plain line-based (TTY) streams. +func DecodeLogStream(r io.Reader, onEntry func(Entry) error) error { + br := bufio.NewReader(r) + header, err := br.Peek(8) + if err != nil { + if err == io.EOF { + return nil + } + return fmt.Errorf("peek log header: %w", err) + } + + // Multiplexed streams start with stream type 0/1/2 and a big-endian size. + if isLikelyMultiplexed(header) { + return decodeMultiplexed(br, onEntry) + } + return decodePlainLines(br, "stdout", onEntry) +} + +func isLikelyMultiplexed(header []byte) bool { + if len(header) < 8 { + return false + } + streamType := header[0] + if streamType > 2 { + return false + } + // Bytes 1-3 are reserved zeroes in the Docker multiplex header. + if header[1] != 0 || header[2] != 0 || header[3] != 0 { + return false + } + return true +} + +func decodeMultiplexed(br *bufio.Reader, onEntry func(Entry) error) error { + header := make([]byte, 8) + for { + if _, err := io.ReadFull(br, header); err != nil { + if err == io.EOF || err == io.ErrUnexpectedEOF { + return nil + } + return fmt.Errorf("read multiplex header: %w", err) + } + + streamName := streamName(header[0]) + size := binary.BigEndian.Uint32(header[4:8]) + if size == 0 { + continue + } + + payload := make([]byte, size) + if _, err := io.ReadFull(br, payload); err != nil { + if err == io.EOF || err == io.ErrUnexpectedEOF { + return nil + } + return fmt.Errorf("read multiplex payload: %w", err) + } + + if err := emitLines(string(payload), streamName, onEntry); err != nil { + return err + } + } +} + +func decodePlainLines(br *bufio.Reader, stream string, onEntry func(Entry) error) error { + for { + line, err := br.ReadString('\n') + if len(line) > 0 { + if err := emitLines(line, stream, onEntry); err != nil { + return err + } + } + if err != nil { + if err == io.EOF { + return nil + } + return fmt.Errorf("read plain log line: %w", err) + } + } +} + +func emitLines(payload string, stream string, onEntry func(Entry) error) error { + payload = strings.ReplaceAll(payload, "\r\n", "\n") + parts := strings.Split(payload, "\n") + for i, part := range parts { + // Keep empty trailing segment only when payload did not end with newline? + // Drop empty fragments from trailing newlines. + if part == "" && i == len(parts)-1 { + continue + } + entry := ParseTimestampedLine(part, stream) + if err := onEntry(entry); err != nil { + return err + } + } + return nil +} + +// ParseTimestampedLine splits a Docker `--timestamps` log line into timestamp + message. +func ParseTimestampedLine(line string, stream string) Entry { + line = strings.TrimRight(line, "\r\n") + timestamp := time.Now().UTC().Format(time.RFC3339Nano) + message := line + + space := strings.IndexByte(line, ' ') + if space > 0 { + candidate := line[:space] + if _, err := time.Parse(time.RFC3339Nano, candidate); err == nil { + timestamp = candidate + message = line[space+1:] + } else if t, err := time.Parse(time.RFC3339, candidate); err == nil { + timestamp = t.UTC().Format(time.RFC3339Nano) + message = line[space+1:] + } + } + + return Entry{ + Timestamp: timestamp, + Stream: stream, + Message: message, + } +} + +func streamName(code byte) string { + switch code { + case streamStdout: + return "stdout" + case streamStderr: + return "stderr" + default: + return "stdout" + } +} diff --git a/apps/agent/internal/logs/decode_test.go b/apps/agent/internal/logs/decode_test.go index 607dc71..92d9d65 100644 --- a/apps/agent/internal/logs/decode_test.go +++ b/apps/agent/internal/logs/decode_test.go @@ -1,87 +1,87 @@ -package logs - -import ( - "bytes" - "encoding/binary" - "io" - "testing" - "time" -) - -func TestParseTimestampedLine(t *testing.T) { - entry := ParseTimestampedLine("2026-07-25T10:00:00.123456789Z Application started", "stdout") - if entry.Timestamp != "2026-07-25T10:00:00.123456789Z" { - t.Fatalf("timestamp=%q", entry.Timestamp) - } - if entry.Message != "Application started" { - t.Fatalf("message=%q", entry.Message) - } - if entry.Stream != "stdout" { - t.Fatalf("stream=%q", entry.Stream) - } -} - -func TestDecodeMultiplexedStdoutStderr(t *testing.T) { - var buf bytes.Buffer - writeFrame(&buf, 1, []byte("2026-07-25T10:00:00Z hello stdout\n")) - writeFrame(&buf, 2, []byte("2026-07-25T10:00:01Z hello stderr\n")) - - var entries []Entry - if err := DecodeLogStream(&buf, func(entry Entry) error { - entries = append(entries, entry) - return nil - }); err != nil { - t.Fatalf("decode: %v", err) - } - - if len(entries) != 2 { - t.Fatalf("entries=%d", len(entries)) - } - if entries[0].Stream != "stdout" || entries[0].Message != "hello stdout" { - t.Fatalf("entry0=%+v", entries[0]) - } - if entries[1].Stream != "stderr" || entries[1].Message != "hello stderr" { - t.Fatalf("entry1=%+v", entries[1]) - } -} - -func TestDecodePlainLines(t *testing.T) { - input := "2026-07-25T10:00:00Z line-one\n2026-07-25T10:00:01Z line-two\n" - var entries []Entry - if err := DecodeLogStream(bytes.NewBufferString(input), func(entry Entry) error { - entries = append(entries, entry) - return nil - }); err != nil { - t.Fatalf("decode: %v", err) - } - if len(entries) != 2 { - t.Fatalf("entries=%d %+v", len(entries), entries) - } - if entries[0].Message != "line-one" || entries[1].Message != "line-two" { - t.Fatalf("messages=%q %q", entries[0].Message, entries[1].Message) - } -} - -func writeFrame(w io.Writer, stream byte, payload []byte) { - var header [8]byte - header[0] = stream - binary.BigEndian.PutUint32(header[4:], uint32(len(payload))) - _, _ = w.Write(header[:]) - _, _ = w.Write(payload) -} - -func TestParseTimestampedLineFallback(t *testing.T) { - before := time.Now().UTC().Add(-time.Second) - entry := ParseTimestampedLine("no-timestamp message", "stderr") - after := time.Now().UTC().Add(time.Second) - if entry.Message != "no-timestamp message" { - t.Fatalf("message=%q", entry.Message) - } - ts, err := time.Parse(time.RFC3339Nano, entry.Timestamp) - if err != nil { - t.Fatalf("parse ts: %v", err) - } - if ts.Before(before) || ts.After(after) { - t.Fatalf("unexpected fallback timestamp %s", entry.Timestamp) - } -} +package logs + +import ( + "bytes" + "encoding/binary" + "io" + "testing" + "time" +) + +func TestParseTimestampedLine(t *testing.T) { + entry := ParseTimestampedLine("2026-07-25T10:00:00.123456789Z Application started", "stdout") + if entry.Timestamp != "2026-07-25T10:00:00.123456789Z" { + t.Fatalf("timestamp=%q", entry.Timestamp) + } + if entry.Message != "Application started" { + t.Fatalf("message=%q", entry.Message) + } + if entry.Stream != "stdout" { + t.Fatalf("stream=%q", entry.Stream) + } +} + +func TestDecodeMultiplexedStdoutStderr(t *testing.T) { + var buf bytes.Buffer + writeFrame(&buf, 1, []byte("2026-07-25T10:00:00Z hello stdout\n")) + writeFrame(&buf, 2, []byte("2026-07-25T10:00:01Z hello stderr\n")) + + var entries []Entry + if err := DecodeLogStream(&buf, func(entry Entry) error { + entries = append(entries, entry) + return nil + }); err != nil { + t.Fatalf("decode: %v", err) + } + + if len(entries) != 2 { + t.Fatalf("entries=%d", len(entries)) + } + if entries[0].Stream != "stdout" || entries[0].Message != "hello stdout" { + t.Fatalf("entry0=%+v", entries[0]) + } + if entries[1].Stream != "stderr" || entries[1].Message != "hello stderr" { + t.Fatalf("entry1=%+v", entries[1]) + } +} + +func TestDecodePlainLines(t *testing.T) { + input := "2026-07-25T10:00:00Z line-one\n2026-07-25T10:00:01Z line-two\n" + var entries []Entry + if err := DecodeLogStream(bytes.NewBufferString(input), func(entry Entry) error { + entries = append(entries, entry) + return nil + }); err != nil { + t.Fatalf("decode: %v", err) + } + if len(entries) != 2 { + t.Fatalf("entries=%d %+v", len(entries), entries) + } + if entries[0].Message != "line-one" || entries[1].Message != "line-two" { + t.Fatalf("messages=%q %q", entries[0].Message, entries[1].Message) + } +} + +func writeFrame(w io.Writer, stream byte, payload []byte) { + var header [8]byte + header[0] = stream + binary.BigEndian.PutUint32(header[4:], uint32(len(payload))) + _, _ = w.Write(header[:]) + _, _ = w.Write(payload) +} + +func TestParseTimestampedLineFallback(t *testing.T) { + before := time.Now().UTC().Add(-time.Second) + entry := ParseTimestampedLine("no-timestamp message", "stderr") + after := time.Now().UTC().Add(time.Second) + if entry.Message != "no-timestamp message" { + t.Fatalf("message=%q", entry.Message) + } + ts, err := time.Parse(time.RFC3339Nano, entry.Timestamp) + if err != nil { + t.Fatalf("parse ts: %v", err) + } + if ts.Before(before) || ts.After(after) { + t.Fatalf("unexpected fallback timestamp %s", entry.Timestamp) + } +} diff --git a/apps/agent/internal/logs/service.go b/apps/agent/internal/logs/service.go index 76a8da3..2729eff 100644 --- a/apps/agent/internal/logs/service.go +++ b/apps/agent/internal/logs/service.go @@ -1,145 +1,145 @@ -package logs - -import ( - "context" - "fmt" - "sync" - - "docksight-agent/internal/logger" -) - -// Service manages concurrent container log streams keyed by requestId. -type Service struct { - engine Engine - emitter ChunkEmitter - - mu sync.Mutex - streams map[string]*Stream - rootCtx context.Context - cancel context.CancelFunc -} - -// NewService creates a logs service. Call SetEmitter before Subscribe. -func NewService(engine Engine) *Service { - ctx, cancel := context.WithCancel(context.Background()) - return &Service{ - engine: engine, - streams: make(map[string]*Stream), - rootCtx: ctx, - cancel: cancel, - } -} - -// SetEmitter wires the communication sender used for logs.chunk messages. -func (s *Service) SetEmitter(emitter ChunkEmitter) { - s.mu.Lock() - defer s.mu.Unlock() - s.emitter = emitter -} - -// Subscribe starts (or replaces) a log stream for requestId. -func (s *Service) Subscribe(opts SubscribeOptions) error { - s.mu.Lock() - emitter := s.emitter - engine := s.engine - root := s.rootCtx - existing := s.streams[opts.RequestID] - if existing != nil { - delete(s.streams, opts.RequestID) - } - s.mu.Unlock() - - if existing != nil { - existing.close() - } - - if engine == nil { - return fmt.Errorf("subscribe: docker engine unavailable") - } - if emitter == nil { - return fmt.Errorf("subscribe: chunk emitter is not configured") - } - - stream, err := newStream(root, engine, opts, emitter) - if err != nil { - return err - } - - s.mu.Lock() - if prev, ok := s.streams[opts.RequestID]; ok { - delete(s.streams, opts.RequestID) - s.mu.Unlock() - prev.close() - s.mu.Lock() - } - s.streams[opts.RequestID] = stream - s.mu.Unlock() - - logger.Info("log stream subscribed", - "requestId", opts.RequestID, - "containerId", opts.ContainerID, - "tail", opts.Tail, - "follow", opts.Follow, - ) - return nil -} - -// Unsubscribe stops a single stream by requestId. -func (s *Service) Unsubscribe(requestID string) error { - if requestID == "" { - return fmt.Errorf("unsubscribe: requestId is required") - } - - s.mu.Lock() - stream, ok := s.streams[requestID] - if ok { - delete(s.streams, requestID) - } - s.mu.Unlock() - - if !ok { - return fmt.Errorf("unsubscribe: stream %q not found", requestID) - } - - stream.close() - logger.Info("log stream unsubscribed", "requestId", requestID) - return nil -} - -// UnsubscribeAll stops every active stream (e.g. on WebSocket disconnect). -func (s *Service) UnsubscribeAll() { - s.mu.Lock() - streams := make([]*Stream, 0, len(s.streams)) - for id, stream := range s.streams { - streams = append(streams, stream) - delete(s.streams, id) - } - s.mu.Unlock() - - for _, stream := range streams { - stream.close() - } -} - -// ActiveCount returns the number of managed streams (testing / diagnostics). -func (s *Service) ActiveCount() int { - s.mu.Lock() - defer s.mu.Unlock() - return len(s.streams) -} - -// HasStream reports whether requestId is currently active. -func (s *Service) HasStream(requestID string) bool { - s.mu.Lock() - defer s.mu.Unlock() - _, ok := s.streams[requestID] - return ok -} - -// Close cancels the service root context and all streams. -func (s *Service) Close() { - s.UnsubscribeAll() - if s.cancel != nil { - s.cancel() - } -} +package logs + +import ( + "context" + "fmt" + "sync" + + "docksight-agent/internal/logger" +) + +// Service manages concurrent container log streams keyed by requestId. +type Service struct { + engine Engine + emitter ChunkEmitter + + mu sync.Mutex + streams map[string]*Stream + rootCtx context.Context + cancel context.CancelFunc +} + +// NewService creates a logs service. Call SetEmitter before Subscribe. +func NewService(engine Engine) *Service { + ctx, cancel := context.WithCancel(context.Background()) + return &Service{ + engine: engine, + streams: make(map[string]*Stream), + rootCtx: ctx, + cancel: cancel, + } +} + +// SetEmitter wires the communication sender used for logs.chunk messages. +func (s *Service) SetEmitter(emitter ChunkEmitter) { + s.mu.Lock() + defer s.mu.Unlock() + s.emitter = emitter +} + +// Subscribe starts (or replaces) a log stream for requestId. +func (s *Service) Subscribe(opts SubscribeOptions) error { + s.mu.Lock() + emitter := s.emitter + engine := s.engine + root := s.rootCtx + existing := s.streams[opts.RequestID] + if existing != nil { + delete(s.streams, opts.RequestID) + } + s.mu.Unlock() + + if existing != nil { + existing.close() + } + + if engine == nil { + return fmt.Errorf("subscribe: docker engine unavailable") + } + if emitter == nil { + return fmt.Errorf("subscribe: chunk emitter is not configured") + } + + stream, err := newStream(root, engine, opts, emitter) + if err != nil { + return err + } + + s.mu.Lock() + if prev, ok := s.streams[opts.RequestID]; ok { + delete(s.streams, opts.RequestID) + s.mu.Unlock() + prev.close() + s.mu.Lock() + } + s.streams[opts.RequestID] = stream + s.mu.Unlock() + + logger.Info("log stream subscribed", + "requestId", opts.RequestID, + "containerId", opts.ContainerID, + "tail", opts.Tail, + "follow", opts.Follow, + ) + return nil +} + +// Unsubscribe stops a single stream by requestId. +func (s *Service) Unsubscribe(requestID string) error { + if requestID == "" { + return fmt.Errorf("unsubscribe: requestId is required") + } + + s.mu.Lock() + stream, ok := s.streams[requestID] + if ok { + delete(s.streams, requestID) + } + s.mu.Unlock() + + if !ok { + return fmt.Errorf("unsubscribe: stream %q not found", requestID) + } + + stream.close() + logger.Info("log stream unsubscribed", "requestId", requestID) + return nil +} + +// UnsubscribeAll stops every active stream (e.g. on WebSocket disconnect). +func (s *Service) UnsubscribeAll() { + s.mu.Lock() + streams := make([]*Stream, 0, len(s.streams)) + for id, stream := range s.streams { + streams = append(streams, stream) + delete(s.streams, id) + } + s.mu.Unlock() + + for _, stream := range streams { + stream.close() + } +} + +// ActiveCount returns the number of managed streams (testing / diagnostics). +func (s *Service) ActiveCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.streams) +} + +// HasStream reports whether requestId is currently active. +func (s *Service) HasStream(requestID string) bool { + s.mu.Lock() + defer s.mu.Unlock() + _, ok := s.streams[requestID] + return ok +} + +// Close cancels the service root context and all streams. +func (s *Service) Close() { + s.UnsubscribeAll() + if s.cancel != nil { + s.cancel() + } +} diff --git a/apps/agent/internal/logs/service_test.go b/apps/agent/internal/logs/service_test.go index 9cd7b28..3ba21c4 100644 --- a/apps/agent/internal/logs/service_test.go +++ b/apps/agent/internal/logs/service_test.go @@ -1,178 +1,178 @@ -package logs - -import ( - "context" - "io" - "strings" - "sync" - "testing" - "time" -) - -type fakeEngine struct { - mu sync.Mutex - readers map[string]*fakeReader -} - -type fakeReader struct { - io.Reader - closed bool - closeCh chan struct{} -} - -func (r *fakeReader) Close() error { - if !r.closed { - r.closed = true - close(r.closeCh) - } - return nil -} - -func newFakeEngine() *fakeEngine { - return &fakeEngine{readers: make(map[string]*fakeReader)} -} - -func (e *fakeEngine) ContainerLogs( - ctx context.Context, - containerID string, - tail string, - follow bool, -) (io.ReadCloser, error) { - _ = ctx - _ = tail - _ = follow - body := "2026-07-25T10:00:00Z boot " + containerID + "\n" - reader := &fakeReader{ - Reader: strings.NewReader(body), - closeCh: make(chan struct{}), - } - e.mu.Lock() - e.readers[containerID] = reader - e.mu.Unlock() - return reader, nil -} - -type recordingEmitter struct { - mu sync.Mutex - chunks []Chunk -} - -func (e *recordingEmitter) EmitLogChunk(chunk Chunk) error { - e.mu.Lock() - defer e.mu.Unlock() - copied := Chunk{ - RequestID: chunk.RequestID, - ContainerID: chunk.ContainerID, - Entries: append([]Entry(nil), chunk.Entries...), - } - e.chunks = append(e.chunks, copied) - return nil -} - -func (e *recordingEmitter) count() int { - e.mu.Lock() - defer e.mu.Unlock() - return len(e.chunks) -} - -func TestSubscribeCreatesStream(t *testing.T) { - engine := newFakeEngine() - emitter := &recordingEmitter{} - svc := NewService(engine) - svc.SetEmitter(emitter) - defer svc.Close() - - err := svc.Subscribe(SubscribeOptions{ - RequestID: "req-1", - ContainerID: "backend", - Tail: 50, - Follow: false, - }) - if err != nil { - t.Fatalf("subscribe: %v", err) - } - if !svc.HasStream("req-1") { - t.Fatal("expected active stream") - } - if svc.ActiveCount() != 1 { - t.Fatalf("active=%d", svc.ActiveCount()) - } - - waitFor(t, 2*time.Second, func() bool { return emitter.count() >= 1 }) -} - -func TestUnsubscribeCancelsOnlyTargetStream(t *testing.T) { - engine := newFakeEngine() - emitter := &recordingEmitter{} - svc := NewService(engine) - svc.SetEmitter(emitter) - defer svc.Close() - - if err := svc.Subscribe(SubscribeOptions{RequestID: "req-1", ContainerID: "backend", Follow: true}); err != nil { - t.Fatalf("subscribe 1: %v", err) - } - if err := svc.Subscribe(SubscribeOptions{RequestID: "req-2", ContainerID: "postgres", Follow: true}); err != nil { - t.Fatalf("subscribe 2: %v", err) - } - if svc.ActiveCount() != 2 { - t.Fatalf("active=%d", svc.ActiveCount()) - } - - if err := svc.Unsubscribe("req-1"); err != nil { - t.Fatalf("unsubscribe: %v", err) - } - if svc.HasStream("req-1") { - t.Fatal("req-1 should be gone") - } - if !svc.HasStream("req-2") { - t.Fatal("req-2 should remain") - } - if svc.ActiveCount() != 1 { - t.Fatalf("active=%d", svc.ActiveCount()) - } -} - -func TestMultipleStreamsIsolation(t *testing.T) { - engine := newFakeEngine() - emitter := &recordingEmitter{} - svc := NewService(engine) - svc.SetEmitter(emitter) - defer svc.Close() - - if err := svc.Subscribe(SubscribeOptions{RequestID: "req-a", ContainerID: "backend", Follow: false}); err != nil { - t.Fatalf("subscribe a: %v", err) - } - if err := svc.Subscribe(SubscribeOptions{RequestID: "req-b", ContainerID: "postgres", Follow: false}); err != nil { - t.Fatalf("subscribe b: %v", err) - } - - waitFor(t, 2*time.Second, func() bool { return emitter.count() >= 2 }) - - emitter.mu.Lock() - defer emitter.mu.Unlock() - seen := map[string]bool{} - for _, chunk := range emitter.chunks { - seen[chunk.RequestID] = true - if chunk.RequestID == "req-a" && chunk.ContainerID != "backend" { - t.Fatalf("req-a mapped to %s", chunk.ContainerID) - } - if chunk.RequestID == "req-b" && chunk.ContainerID != "postgres" { - t.Fatalf("req-b mapped to %s", chunk.ContainerID) - } - } - if !seen["req-a"] || !seen["req-b"] { - t.Fatalf("seen=%v", seen) - } -} - -func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { - t.Helper() - deadline := time.Now().Add(timeout) - for time.Now().Before(deadline) { - if cond() { - return - } - time.Sleep(20 * time.Millisecond) - } - t.Fatal("condition not met before timeout") -} +package logs + +import ( + "context" + "io" + "strings" + "sync" + "testing" + "time" +) + +type fakeEngine struct { + mu sync.Mutex + readers map[string]*fakeReader +} + +type fakeReader struct { + io.Reader + closed bool + closeCh chan struct{} +} + +func (r *fakeReader) Close() error { + if !r.closed { + r.closed = true + close(r.closeCh) + } + return nil +} + +func newFakeEngine() *fakeEngine { + return &fakeEngine{readers: make(map[string]*fakeReader)} +} + +func (e *fakeEngine) ContainerLogs( + ctx context.Context, + containerID string, + tail string, + follow bool, +) (io.ReadCloser, error) { + _ = ctx + _ = tail + _ = follow + body := "2026-07-25T10:00:00Z boot " + containerID + "\n" + reader := &fakeReader{ + Reader: strings.NewReader(body), + closeCh: make(chan struct{}), + } + e.mu.Lock() + e.readers[containerID] = reader + e.mu.Unlock() + return reader, nil +} + +type recordingEmitter struct { + mu sync.Mutex + chunks []Chunk +} + +func (e *recordingEmitter) EmitLogChunk(chunk Chunk) error { + e.mu.Lock() + defer e.mu.Unlock() + copied := Chunk{ + RequestID: chunk.RequestID, + ContainerID: chunk.ContainerID, + Entries: append([]Entry(nil), chunk.Entries...), + } + e.chunks = append(e.chunks, copied) + return nil +} + +func (e *recordingEmitter) count() int { + e.mu.Lock() + defer e.mu.Unlock() + return len(e.chunks) +} + +func TestSubscribeCreatesStream(t *testing.T) { + engine := newFakeEngine() + emitter := &recordingEmitter{} + svc := NewService(engine) + svc.SetEmitter(emitter) + defer svc.Close() + + err := svc.Subscribe(SubscribeOptions{ + RequestID: "req-1", + ContainerID: "backend", + Tail: 50, + Follow: false, + }) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + if !svc.HasStream("req-1") { + t.Fatal("expected active stream") + } + if svc.ActiveCount() != 1 { + t.Fatalf("active=%d", svc.ActiveCount()) + } + + waitFor(t, 2*time.Second, func() bool { return emitter.count() >= 1 }) +} + +func TestUnsubscribeCancelsOnlyTargetStream(t *testing.T) { + engine := newFakeEngine() + emitter := &recordingEmitter{} + svc := NewService(engine) + svc.SetEmitter(emitter) + defer svc.Close() + + if err := svc.Subscribe(SubscribeOptions{RequestID: "req-1", ContainerID: "backend", Follow: true}); err != nil { + t.Fatalf("subscribe 1: %v", err) + } + if err := svc.Subscribe(SubscribeOptions{RequestID: "req-2", ContainerID: "postgres", Follow: true}); err != nil { + t.Fatalf("subscribe 2: %v", err) + } + if svc.ActiveCount() != 2 { + t.Fatalf("active=%d", svc.ActiveCount()) + } + + if err := svc.Unsubscribe("req-1"); err != nil { + t.Fatalf("unsubscribe: %v", err) + } + if svc.HasStream("req-1") { + t.Fatal("req-1 should be gone") + } + if !svc.HasStream("req-2") { + t.Fatal("req-2 should remain") + } + if svc.ActiveCount() != 1 { + t.Fatalf("active=%d", svc.ActiveCount()) + } +} + +func TestMultipleStreamsIsolation(t *testing.T) { + engine := newFakeEngine() + emitter := &recordingEmitter{} + svc := NewService(engine) + svc.SetEmitter(emitter) + defer svc.Close() + + if err := svc.Subscribe(SubscribeOptions{RequestID: "req-a", ContainerID: "backend", Follow: false}); err != nil { + t.Fatalf("subscribe a: %v", err) + } + if err := svc.Subscribe(SubscribeOptions{RequestID: "req-b", ContainerID: "postgres", Follow: false}); err != nil { + t.Fatalf("subscribe b: %v", err) + } + + waitFor(t, 2*time.Second, func() bool { return emitter.count() >= 2 }) + + emitter.mu.Lock() + defer emitter.mu.Unlock() + seen := map[string]bool{} + for _, chunk := range emitter.chunks { + seen[chunk.RequestID] = true + if chunk.RequestID == "req-a" && chunk.ContainerID != "backend" { + t.Fatalf("req-a mapped to %s", chunk.ContainerID) + } + if chunk.RequestID == "req-b" && chunk.ContainerID != "postgres" { + t.Fatalf("req-b mapped to %s", chunk.ContainerID) + } + } + if !seen["req-a"] || !seen["req-b"] { + t.Fatalf("seen=%v", seen) + } +} + +func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("condition not met before timeout") +} diff --git a/apps/agent/internal/logs/stream.go b/apps/agent/internal/logs/stream.go index 729187c..303e8ca 100644 --- a/apps/agent/internal/logs/stream.go +++ b/apps/agent/internal/logs/stream.go @@ -1,166 +1,166 @@ -package logs - -import ( - "context" - "fmt" - "io" - "sync" - "time" - - "docksight-agent/internal/logger" -) - -const ( - defaultBatchSize = 50 - defaultBatchInterval = 200 * time.Millisecond -) - -// Stream is one active container log subscription keyed by requestId. -type Stream struct { - RequestID string - ContainerID string - - cancel context.CancelFunc - reader io.ReadCloser - - mu sync.Mutex - closed bool -} - -func (s *Stream) close() { - s.mu.Lock() - defer s.mu.Unlock() - if s.closed { - return - } - s.closed = true - if s.cancel != nil { - s.cancel() - } - if s.reader != nil { - _ = s.reader.Close() - } -} - -func (s *Stream) run( - ctx context.Context, - emit ChunkEmitter, - batchSize int, - batchInterval time.Duration, -) { - defer s.close() - - if batchSize <= 0 { - batchSize = defaultBatchSize - } - if batchInterval <= 0 { - batchInterval = defaultBatchInterval - } - - entries := make(chan Entry, batchSize*2) - errCh := make(chan error, 1) - - go func() { - defer close(entries) - err := DecodeLogStream(s.reader, func(entry Entry) error { - select { - case <-ctx.Done(): - return ctx.Err() - case entries <- entry: - return nil - } - }) - if err != nil && ctx.Err() == nil { - errCh <- err - } - }() - - batch := make([]Entry, 0, batchSize) - ticker := time.NewTicker(batchInterval) - defer ticker.Stop() - - flush := func() { - if len(batch) == 0 || emit == nil { - batch = batch[:0] - return - } - chunk := Chunk{ - RequestID: s.RequestID, - ContainerID: s.ContainerID, - Entries: append([]Entry(nil), batch...), - } - batch = batch[:0] - if err := emit.EmitLogChunk(chunk); err != nil { - logger.Warn("emit log chunk failed", - "requestId", s.RequestID, - "error", err.Error(), - ) - } - } - - for { - select { - case <-ctx.Done(): - flush() - return - case err := <-errCh: - flush() - if err != nil { - logger.Warn("log stream ended with error", - "requestId", s.RequestID, - "containerId", s.ContainerID, - "error", err.Error(), - ) - } - return - case entry, ok := <-entries: - if !ok { - flush() - return - } - batch = append(batch, entry) - if len(batch) >= batchSize { - flush() - } - case <-ticker.C: - flush() - } - } -} - -// newStream opens Docker logs and starts the decode/batch goroutine. -func newStream( - parent context.Context, - engine Engine, - opts SubscribeOptions, - emit ChunkEmitter, -) (*Stream, error) { - if opts.RequestID == "" { - return nil, fmt.Errorf("subscribe: requestId is required") - } - if opts.ContainerID == "" { - return nil, fmt.Errorf("subscribe: containerId is required") - } - - tail := opts.Tail - if tail <= 0 { - tail = 100 - } - - ctx, cancel := context.WithCancel(parent) - reader, err := engine.ContainerLogs(ctx, opts.ContainerID, fmt.Sprintf("%d", tail), opts.Follow) - if err != nil { - cancel() - return nil, fmt.Errorf("subscribe: %w", err) - } - - stream := &Stream{ - RequestID: opts.RequestID, - ContainerID: opts.ContainerID, - cancel: cancel, - reader: reader, - } - - go stream.run(ctx, emit, defaultBatchSize, defaultBatchInterval) - return stream, nil -} +package logs + +import ( + "context" + "fmt" + "io" + "sync" + "time" + + "docksight-agent/internal/logger" +) + +const ( + defaultBatchSize = 50 + defaultBatchInterval = 200 * time.Millisecond +) + +// Stream is one active container log subscription keyed by requestId. +type Stream struct { + RequestID string + ContainerID string + + cancel context.CancelFunc + reader io.ReadCloser + + mu sync.Mutex + closed bool +} + +func (s *Stream) close() { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + s.closed = true + if s.cancel != nil { + s.cancel() + } + if s.reader != nil { + _ = s.reader.Close() + } +} + +func (s *Stream) run( + ctx context.Context, + emit ChunkEmitter, + batchSize int, + batchInterval time.Duration, +) { + defer s.close() + + if batchSize <= 0 { + batchSize = defaultBatchSize + } + if batchInterval <= 0 { + batchInterval = defaultBatchInterval + } + + entries := make(chan Entry, batchSize*2) + errCh := make(chan error, 1) + + go func() { + defer close(entries) + err := DecodeLogStream(s.reader, func(entry Entry) error { + select { + case <-ctx.Done(): + return ctx.Err() + case entries <- entry: + return nil + } + }) + if err != nil && ctx.Err() == nil { + errCh <- err + } + }() + + batch := make([]Entry, 0, batchSize) + ticker := time.NewTicker(batchInterval) + defer ticker.Stop() + + flush := func() { + if len(batch) == 0 || emit == nil { + batch = batch[:0] + return + } + chunk := Chunk{ + RequestID: s.RequestID, + ContainerID: s.ContainerID, + Entries: append([]Entry(nil), batch...), + } + batch = batch[:0] + if err := emit.EmitLogChunk(chunk); err != nil { + logger.Warn("emit log chunk failed", + "requestId", s.RequestID, + "error", err.Error(), + ) + } + } + + for { + select { + case <-ctx.Done(): + flush() + return + case err := <-errCh: + flush() + if err != nil { + logger.Warn("log stream ended with error", + "requestId", s.RequestID, + "containerId", s.ContainerID, + "error", err.Error(), + ) + } + return + case entry, ok := <-entries: + if !ok { + flush() + return + } + batch = append(batch, entry) + if len(batch) >= batchSize { + flush() + } + case <-ticker.C: + flush() + } + } +} + +// newStream opens Docker logs and starts the decode/batch goroutine. +func newStream( + parent context.Context, + engine Engine, + opts SubscribeOptions, + emit ChunkEmitter, +) (*Stream, error) { + if opts.RequestID == "" { + return nil, fmt.Errorf("subscribe: requestId is required") + } + if opts.ContainerID == "" { + return nil, fmt.Errorf("subscribe: containerId is required") + } + + tail := opts.Tail + if tail <= 0 { + tail = 100 + } + + ctx, cancel := context.WithCancel(parent) + reader, err := engine.ContainerLogs(ctx, opts.ContainerID, fmt.Sprintf("%d", tail), opts.Follow) + if err != nil { + cancel() + return nil, fmt.Errorf("subscribe: %w", err) + } + + stream := &Stream{ + RequestID: opts.RequestID, + ContainerID: opts.ContainerID, + cancel: cancel, + reader: reader, + } + + go stream.run(ctx, emit, defaultBatchSize, defaultBatchInterval) + return stream, nil +} diff --git a/apps/agent/internal/logs/types.go b/apps/agent/internal/logs/types.go index cdfc104..2d1e2b1 100644 --- a/apps/agent/internal/logs/types.go +++ b/apps/agent/internal/logs/types.go @@ -1,38 +1,38 @@ -package logs - -import ( - "context" - "io" -) - -// Entry is one decoded container log line. -type Entry struct { - Timestamp string `json:"timestamp"` - Stream string `json:"stream"` - Message string `json:"message"` -} - -// SubscribeOptions configures a log stream from the protocol payload. -type SubscribeOptions struct { - RequestID string - ContainerID string - Tail int - Follow bool -} - -// Chunk is a batched set of log entries ready to send as logs.chunk. -type Chunk struct { - RequestID string - ContainerID string - Entries []Entry -} - -// ChunkEmitter sends batched log chunks to the DockSight server. -type ChunkEmitter interface { - EmitLogChunk(chunk Chunk) error -} - -// Engine is the Docker log source used by the logs service. -type Engine interface { - ContainerLogs(ctx context.Context, containerID string, tail string, follow bool) (io.ReadCloser, error) -} +package logs + +import ( + "context" + "io" +) + +// Entry is one decoded container log line. +type Entry struct { + Timestamp string `json:"timestamp"` + Stream string `json:"stream"` + Message string `json:"message"` +} + +// SubscribeOptions configures a log stream from the protocol payload. +type SubscribeOptions struct { + RequestID string + ContainerID string + Tail int + Follow bool +} + +// Chunk is a batched set of log entries ready to send as logs.chunk. +type Chunk struct { + RequestID string + ContainerID string + Entries []Entry +} + +// ChunkEmitter sends batched log chunks to the DockSight server. +type ChunkEmitter interface { + EmitLogChunk(chunk Chunk) error +} + +// Engine is the Docker log source used by the logs service. +type Engine interface { + ContainerLogs(ctx context.Context, containerID string, tail string, follow bool) (io.ReadCloser, error) +} diff --git a/apps/cli/cmd/internal/filesystem/directories.go b/apps/cli/cmd/internal/filesystem/directories.go index 1b809d1..78c729d 100644 --- a/apps/cli/cmd/internal/filesystem/directories.go +++ b/apps/cli/cmd/internal/filesystem/directories.go @@ -1,38 +1,37 @@ -package filesystem - -import ( - "os" -) - -// TempWorkspace creates a private staging directory for release artifacts, -// unique to this run and readable only by the current user. Unlike a fixed -// path under /tmp it cannot collide with files left by another user. -func TempWorkspace() (string, error) { - return os.MkdirTemp("", "docksight-") -} - - -func CreateDirectories( - installationDir string, - dataDir string, -) error { - - directories := []string{ - installationDir, - dataDir, - } - - for _, dir := range directories { - - err := os.MkdirAll( - dir, - 0755, - ) - - if err != nil { - return err - } - } - - return nil -} \ No newline at end of file +package filesystem + +import ( + "os" +) + +// TempWorkspace creates a private staging directory for release artifacts, +// unique to this run and readable only by the current user. Unlike a fixed +// path under /tmp it cannot collide with files left by another user. +func TempWorkspace() (string, error) { + return os.MkdirTemp("", "docksight-") +} + +func CreateDirectories( + installationDir string, + dataDir string, +) error { + + directories := []string{ + installationDir, + dataDir, + } + + for _, dir := range directories { + + err := os.MkdirAll( + dir, + 0755, + ) + + if err != nil { + return err + } + } + + return nil +} diff --git a/apps/cli/cmd/internal/ui/banner.go b/apps/cli/cmd/internal/ui/banner.go index 2efe72c..76ab3c0 100644 --- a/apps/cli/cmd/internal/ui/banner.go +++ b/apps/cli/cmd/internal/ui/banner.go @@ -1,16 +1,16 @@ -package ui - -import "fmt" - -func Banner() { - - fmt.Println(` - ____ _ ____ _ _ -| _ \ ___ ___| | _/ ___|(_) __ _| |__ -| | | |/ _ \ / __| |/ /\___ \| |/ _` + "`" + ` | '_ \ -| |_| | (_) | (__| < ___) | | (_| | | | | -|____/ \___/ \___|_|\_\|____/|_|\__, |_| |_| - |___/`) - fmt.Println("Container monitoring platform") - fmt.Println() -} \ No newline at end of file +package ui + +import "fmt" + +func Banner() { + + fmt.Println(` + ____ _ ____ _ _ +| _ \ ___ ___| | _/ ___|(_) __ _| |__ +| | | |/ _ \ / __| |/ /\___ \| |/ _` + "`" + ` | '_ \ +| |_| | (_) | (__| < ___) | | (_| | | | | +|____/ \___/ \___|_|\_\|____/|_|\__, |_| |_| + |___/`) + fmt.Println("Container monitoring platform") + fmt.Println() +} diff --git a/apps/cli/main.go b/apps/cli/main.go index 4d19b6a..77bdb53 100644 --- a/apps/cli/main.go +++ b/apps/cli/main.go @@ -1,9 +1,9 @@ -package main - -import ( - "github.com/Open-Source-Kigali/docksight/apps/cli/cmd" -) - -func main() { - cmd.Execute() -} \ No newline at end of file +package main + +import ( + "github.com/Open-Source-Kigali/docksight/apps/cli/cmd" +) + +func main() { + cmd.Execute() +}