Add Support for Fluent-bit

agent/Dockerfile.fluentbit

@@ -0,0 +1,51 @@
+# Use the official Golang image to build the binary.
+FROM golang:1.23-alpine AS builder
+
+# Set the Current Working Directory inside the container
+WORKDIR /app
+
+# Copy go mod and sum files
+COPY go.mod go.sum ./
+RUN go mod download
+
+# Copy the source code.
+COPY . .
+
+# Build the Go app. Replace "myapp" with your binary name.
+RUN go build -o ctrlb-agent ./cmd/ctrlb_collector/main.go
+
+# Get Fluent Bit binary
+FROM fluent/fluent-bit:3.2 AS fluentbit
+
+# Use Debian slim as base for compatibility with Fluent Bit
+FROM debian:bookworm-slim
+
+# Install required runtime dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ca-certificates \
+    libyaml-0-2 \
+    libsystemd0 \
+    libcurl4 \
+    libsasl2-2 \
+    libpq5 \
+    libgcc-s1 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Create directory structure
+RUN mkdir -p /opt/fluent-bit/bin
+
+# Copy Fluent Bit binary and libraries from official image
+COPY --from=fluentbit /fluent-bit /opt/fluent-bit
+
+# Add Fluent Bit to PATH
+ENV PATH="/opt/fluent-bit/bin:${PATH}"
+
+WORKDIR /app
+
+# Copy the binary from the builder stage.
+COPY --from=builder /app/ctrlb-agent .
+COPY --from=builder /app/internal/config/fluentbit-config.yaml ./config.yaml
+
+# Command to run the executable
+CMD ["./ctrlb-agent"]
+

agent/README.md

@@ -1,6 +1,6 @@
 # 🛰️ CTRLB Agent
 
-The **CTRLB Agent** is a lightweight wrapper built on top of the OpenTelemetry Collector. Once installed, it connects to the CTRLB backend, shares its runtime status, and receives its initial configuration. The agent can be managed remotely through a set of defined HTTP endpoints.
+The **CTRLB Agent** is a flexible telemetry collection agent that supports multiple backends including **OpenTelemetry Collector** and **Fluent Bit**. Once installed, it connects to the CTRLB backend, shares its runtime status, and receives its initial configuration. The agent can be managed remotely through a set of defined HTTP endpoints.
 
 ---
 
@@ -17,10 +17,92 @@ The **CTRLB Agent** is a lightweight wrapper built on top of the OpenTelemetry C
 
 The backend provides a platform-specific installation script. Once executed, the agent starts and reaches out to the backend for initial configuration.
 
-You can customize agent behavior using the following command-line flags:
+You can customize agent behavior using environment variables:
 
-- `--config`: Path to the agent configuration file. Default is `./config.yaml`
-- `--backend`: URL of the backend server. Default is `http://pipeline.ctrlb.ai:8096`
+- `AGENT_TYPE`: Type of telemetry backend to use (`otel`, `fluent-bit`, `fluentbit`, `fb`). Default is `otel`
+- `AGENT_CONFIG_PATH`: Path to the agent configuration file. Default is `./config.yaml`
+- `BACKEND_URL`: URL of the backend server. Default is `http://pipeline.ctrlb.ai:8096`
+- `PIPELINE_NAME`: Name of the pipeline (optional)
+- `STARTED_BY`: User who started the agent (optional)
+
+### Supported Backends
+
+#### OpenTelemetry Collector (Default)
+The OpenTelemetry Collector runs as an embedded library within the agent process.
+
+```bash
+export AGENT_TYPE=otel
+export AGENT_CONFIG_PATH=./otel-config.yaml
+./ctrlb_collector
+```
+
+#### Fluent Bit
+Fluent Bit runs as a child process controlled by the agent via HTTP APIs.
+
+```bash
+export AGENT_TYPE=fluent-bit
+export AGENT_CONFIG_PATH=./fluentbit-config.yaml
+./ctrlb_collector
+```
+
+See [Fluent Bit Integration Guide](../docs/collector/fluentbit-integration.md) for detailed setup instructions.
+
+---
+
+## 🐳 Docker Deployment
+
+### Building the Fluent Bit Agent Image
+
+Build the Docker image with Fluent Bit support:
+
+```bash
+docker build -f Dockerfile.fluentbit -t ctrlb-agent-fluentbit:latest .
+```
+
+### Running the Fluent Bit Agent Container
+
+Run the agent container with Fluent Bit:
+
+```bash
+docker run -d \
+  --name ctrlb-agent-fluentbit \
+  -p 3421:3421 \
+  -v $(pwd)/internal/config:/app/internal/config \
+  -v /var/log:/var/log \
+  -e BACKEND_URL=http://host.docker.internal:8096 \
+  -e AGENT_TYPE=fluent-bit \
+  -e AGENT_CONFIG_PATH=./internal/config/config.yaml \
+  -e PORT=3421 \
+  ctrlb-agent-fluentbit:latest
+```
+
+**Configuration Options:**
+
+- `-p 3421:3421` – Exposes the agent API port
+- `-v $(pwd)/internal/config:/app/internal/config` – Mounts config directory for dynamic updates
+- `-v /var/log:/var/log` – Mounts host logs for collection (optional)
+- `-e BACKEND_URL` – Backend server URL (use `host.docker.internal` for local development)
+- `-e AGENT_TYPE=fluent-bit` – Specifies Fluent Bit as the telemetry backend
+- `-e AGENT_CONFIG_PATH` – Path to Fluent Bit configuration file
+- `-e PORT=3421` – Agent API port
+
+**Useful Commands:**
+
+```bash
+# View agent logs
+docker logs -f ctrlb-agent-fluentbit
+
+# Stop the agent
+docker stop ctrlb-agent-fluentbit
+
+# Remove the container
+docker rm ctrlb-agent-fluentbit
+
+# Interactive mode (for testing)
+docker run -it --rm \
+  -e BACKEND_URL=http://host.docker.internal:8096 \
+  ctrlb-agent-fluentbit:latest
+```
 
 ---
 
@@ -32,9 +114,9 @@ All endpoints are served under the base path: `/agent/v1`
 
 These endpoints control the agent's operational state:
 
-- `POST /agent/v1/start` – Start the OpenTelemetry Collector instance without restarting the agent process.
-- `POST /agent/v1/stop` – Stop the OpenTelemetry Collector instance while keeping the agent process alive.
-- `POST /agent/v1/shutdown` – Gracefully shut down the agent.
+- `POST /agent/v1/start` – Start the telemetry collector (OTEL/Fluent Bit) without restarting the agent process.
+- `POST /agent/v1/stop` – Stop the telemetry collector while keeping the agent process alive.
+- `POST /agent/v1/shutdown` – Gracefully shut down the agent and collector.
 
 ### Configuration
 
@@ -49,9 +131,16 @@ These endpoints manage the agent’s configuration:
 ## 🛠️ Tech Stack
 
 - **Go** – Core implementation language.
-- **OpenTelemetry Collector** – Base collector engine.
+- **OpenTelemetry Collector** – Embedded telemetry collector (default).
+- **Fluent Bit** – Lightweight log processor (alternative backend).
 - **HTTP + Gorilla Mux** – Communication protocol and routing.
 
+## 📚 Documentation
+
+- [Fluent Bit Integration Guide](../docs/collector/fluentbit-integration.md)
+- [Fluent Bit Quick Start](../docs/collector/fluentbit-quickstart.md)
+- [Architecture Overview](../docs/architecture.md)
+
 ---
 
 ## 📄 License

agent/cmd/ctrlb_collector/main.go

@@ -4,6 +4,8 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"slices"
+	"strconv"
 	"sync"
 	"syscall"
 	"time"
@@ -48,6 +50,38 @@ func main() {
 	if constants.STARTED_BY == "" {
 		logger.Logger.Info("STARTED_BY environment variable is not set. Using default value: empty string.")
 	}
+
+	constants.PORT = os.Getenv("PORT")
+	if constants.PORT == "" {
+		logger.Logger.Info("PORT environment variable is not set. Using default value: 3421.")
+		constants.PORT = "3421"
+	}
+
+	constants.AGENT_TYPE = os.Getenv("AGENT_TYPE")
+	if constants.AGENT_TYPE == "" {
+		logger.Logger.Info("AGENT_TYPE environment variable is not set. Using default value: otel.")
+		constants.AGENT_TYPE = "otel"
+	}
+
+	isValid := slices.Contains(constants.SUPPORTED_AGENT_TYPES, constants.AGENT_TYPE)
+	if !isValid {
+		logger.Logger.Sugar().Infof("Invalid AGENT_TYPE: %s. Using default value: otel.", constants.AGENT_TYPE)
+		constants.AGENT_TYPE = "otel"
+	}
+
+	// Configure heartbeat interval
+	if heartbeatIntervalEnv := os.Getenv("HEARTBEAT_INTERVAL_SEC"); heartbeatIntervalEnv != "" {
+		if interval, err := strconv.Atoi(heartbeatIntervalEnv); err == nil && interval > 0 {
+			constants.HEARTBEAT_INTERVAL_SEC = interval
+		}
+	}
+
+	// SKIP_CONFIG_VALIDATION: Optional, skip fluent-bit binary config validation
+	if skipValidation := os.Getenv("SKIP_CONFIG_VALIDATION"); skipValidation == "true" || skipValidation == "1" {
+		constants.SKIP_CONFIG_VALIDATION = true
+		logger.Logger.Info("SKIP_CONFIG_VALIDATION is enabled. Skipping fluent-bit binary config validation.")
+	}
+
 	// Check if config file exists
 	if _, err := os.Stat(constants.AGENT_CONFIG_PATH); err != nil {
 		logger.Logger.Sugar().Errorf("Config file doesn't exist at location: %v", constants.AGENT_CONFIG_PATH)
@@ -70,10 +104,14 @@ func main() {
 
 	version, err := adapter.GetVersion()
 	if err != nil {
-		logger.Logger.Sugar().Fatalf("Error while fetching agent version: %v", err)
-	} else {
-		constants.AGENT_VERSION = version
+		logger.Logger.Sugar().Errorf("Error while fetching agent version: %v", err)
+		adapter.GracefulShutdown()
+		os.Exit(1)
 	}
+	constants.AGENT_VERSION = version
+
+	// Channel to signal fatal errors from goroutines
+	fatalErrChan := make(chan error, 2)
 
 	// Call Backend server which will be informed about agent being started
 	wg.Add(1)
@@ -85,19 +123,26 @@ func main() {
 		}
 		serverStartConfig, err := client.InformBackendServerStart(sys, httpClient)
 		if err != nil {
-			logger.Logger.Sugar().Fatalf("Failed to register with backend server: %v", err)
-		} else {
-			err = config.SaveToYAML(serverStartConfig, constants.AGENT_CONFIG_PATH)
-			if err != nil {
-				logger.Logger.Sugar().Fatalf("Error writing config to file: %v", err)
-			}
-			logger.Logger.Info("Successfully registered with the backend server")
+			logger.Logger.Sugar().Errorf("Failed to register with backend server: %v", err)
+			fatalErrChan <- err
+			return
+		}
+		err = config.SaveToYAML(serverStartConfig, constants.AGENT_CONFIG_PATH)
+		if err != nil {
+			logger.Logger.Sugar().Errorf("Error writing config to file: %v", err)
+			fatalErrChan <- err
+			return
 		}
+		logger.Logger.Info("Successfully registered with the backend server")
 	}()
 
-	operator_service := *operators.NewOperatorService(adapter)
+	// Start heartbeat manager to periodically send status/metrics to backend
+	heartbeatManager := client.NewHeartbeatManager(constants.HEARTBEAT_INTERVAL_SEC)
+	heartbeatManager.Start()
 
-	handler := api.NewRouter(&operator_service)
+	operatorService := operators.NewOperatorService(adapter, constants.AGENT_TYPE)
+
+	handler := api.NewRouter(operatorService)
 
 	server := &http.Server{
 		Addr:    ":" + constants.PORT,
@@ -113,19 +158,25 @@ func main() {
 		logger.Logger.Sugar().Infof("Client started at port: %s", constants.PORT)
 		err := server.ListenAndServe()
 		if err != nil && err != http.ErrServerClosed {
-			logger.Logger.Sugar().Fatalf("Failed to start Server: %v", err)
+			logger.Logger.Sugar().Errorf("Failed to start Server: %v", err)
+			fatalErrChan <- err
 		}
 	}()
 
 	// Set up signal handling for graceful shutdown
 	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
 
-	// Wait for termination signal
-	<-sigChan
-
-	logger.Logger.Info("Received termination signal. Initiating graceful shutdown...")
+	// Wait for termination signal or fatal error
+	select {
+	case <-sigChan:
+		logger.Logger.Info("Received termination signal. Initiating graceful shutdown...")
+	case err := <-fatalErrChan:
+		logger.Logger.Sugar().Errorf("Fatal error occurred: %v. Initiating graceful shutdown...", err)
+	}
 
+	heartbeatManager.Stop()
+	shutdown.ShutdownServer()
 	adapter.GracefulShutdown()
-
+	wg.Wait()
 }

agent/go.mod

@@ -157,8 +157,8 @@ require (
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus-community/windows_exporter v0.27.2 // indirect
 	github.com/prometheus/client_golang v1.21.1 // indirect
-	github.com/prometheus/client_model v0.6.1 // indirect
-	github.com/prometheus/common v0.62.0 // indirect
+	github.com/prometheus/client_model v0.6.1
+	github.com/prometheus/common v0.62.0
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/rs/cors v1.11.1 // indirect

agent/internal/adapters/adapter.go

@@ -14,11 +14,16 @@ type Adapter interface {
 	GracefulShutdown() error
 	GetVersion() (string, error)
 	ValidateConfigInMemory(data *map[string]any) error
+	ValidateConfigOnDisk(data *map[string]any) error
+	GetMetrics() (map[string]any, error)
 }
 
 func NewAdapter(wg *sync.WaitGroup, agentType string) (Adapter, error) {
 	if agentType == "otel" || agentType == "" {
 		return NewOTELAdapter(wg), nil
 	}
+	if agentType == "fluent-bit" || agentType == "fluentbit" || agentType == "fb" {
+		return NewFluentBitAdapter(wg), nil
+	}
 	return nil, fmt.Errorf("unsupported agent type: %s", agentType)
 }

agent/internal/adapters/adapter_test.go

@@ -20,6 +20,16 @@ func TestNewAdapter_OTEL(t *testing.T) {
 	assert.True(t, ok, "adapter should be of type OTELAdapter")
 }
 
+func TestNewAdapter_FLUENTBIT(t *testing.T) {
+	wg := &sync.WaitGroup{}
+
+	adapter, err := adapters.NewAdapter(wg, "fluent-bit")
+	assert.NoError(t, err)
+	assert.NotNil(t, adapter)
+	_, ok := adapter.(*adapters.FluentBitAdapter)
+	assert.True(t, ok, "adapter should be of type FluentBitAdapter")
+}
+
 func TestNewAdapter_DefaultEmptyType(t *testing.T) {
 	wg := &sync.WaitGroup{}
 
@@ -31,7 +41,7 @@ func TestNewAdapter_DefaultEmptyType(t *testing.T) {
 func TestNewAdapter_UnsupportedType(t *testing.T) {
 	wg := &sync.WaitGroup{}
 
-	adapter, err := adapters.NewAdapter(wg, "fluentbit")
+	adapter, err := adapters.NewAdapter(wg, "unsupported-type")
 	assert.Error(t, err)
 	assert.Nil(t, adapter)
 	assert.Contains(t, err.Error(), "unsupported agent type")