more progres

Author: e-n-0

contrib/haproxy/stream-processing-offload/cmd/spoa/.gitignore

@@ -0,0 +1 @@
+spoa

contrib/haproxy/stream-processing-offload/cmd/spoa/Dockerfile

@@ -0,0 +1,35 @@
+# Build stage
+FROM golang:1.24-alpine AS builder
+ENV CGO_ENABLED=1
+
+WORKDIR /app
+COPY . .
+
+RUN apk add --no-cache --update git build-base openssl
+
+# Build the spoa binary
+RUN go build -tags=appsec -o ./contrib/haproxy/stream-processing-offload/cmd/spoa/spoa ./contrib/haproxy/stream-processing-offload/cmd/spoa
+
+# Runtime stage
+FROM alpine:3.20.3
+
+# Set opencontainers labels for Github container registry
+LABEL org.opencontainers.image.source=https://github.com/DataDog/dd-trace-go/tree/main/contrib/haproxy/stream-processing-offload/cmd/spoa
+LABEL org.opencontainers.image.description="An HAProxy Stream Processing Offload Agent service with Datadog App & API Protection support"
+LABEL org.opencontainers.image.licenses=Apache-2.0
+
+ARG COMMIT_SHA=""
+LABEL org.opencontainers.image.revision=${COMMIT_SHA}
+
+RUN apk --no-cache add ca-certificates tzdata libc6-compat libgcc libstdc++
+WORKDIR /app
+
+ARG DD_VERSION=""
+ENV DD_VERSION=${DD_VERSION}
+
+COPY --from=builder /app/contrib/haproxy/stream-processing-offload/cmd/spoa/spoa /app/spoa
+
+EXPOSE 3000
+EXPOSE 3080
+
+CMD ["./spoa"]

contrib/haproxy/stream-processing-offload/cmd/spoa/README.md

@@ -0,0 +1,43 @@
+# HAProxy Stream Processing Offload (SPOA) with Datadog App & API Protection
+
+[GCP Services Extensions](https://cloud.google.com/service-extensions/docs/overview) enable Google Cloud users to provide programmability and extensibility on Cloud Load Balancing data paths and at the edge.
+
+## Installation
+
+### From Release
+
+This package provides a docker image to be used with Google Cloud Service Extensions.
+The images are published at each release of the tracer and can be found in [the repo registry](https://github.com/DataDog/dd-trace-go/pkgs/container/dd-trace-go%2Fservice-extensions-callout).
+
+### Build image
+
+The docker image can be build locally using docker. Start by cloning the `dd-trace-go` repo, `cd` inside it and run that command:
+```sh
+docker build -f contrib/haproxy/stream-processing-offload/cmd/spoa/Dockerfile -t datadog/dd-trace-go/haproxy-spoa:local .
+```
+
+## Configuration
+
+The ASM Service Extension expose some configuration. The configuration can be tweaked if the Service Extension is only used as an External Processor for Envoy that is not operated by GCP.
+
+>**GCP requires that the default configuration for the Service Extension should not change.**
+
+| Environment variable | Default value | Description                                                                                                   |
+|---|---------------|---------------------------------------------------------------------------------------------------------------|
+| `DD_SERVICE_EXTENSION_HOST` | `0.0.0.0`     | Host on where the gRPC and HTTP server should listen to.                                                      |
+| `DD_SERVICE_EXTENSION_PORT` | `443`         | Port used by the gRPC Server.<br>Envoy Google backend’s is only using secure connection to Service Extension. |
+| `DD_SERVICE_EXTENSION_HEALTHCHECK_PORT` | `80`          | Port used for the HTTP server for the health check.                                                           |
+| `DD_SERVICE_EXTENSION_OBSERVABILITY_MODE` | `false`       | Enable observability mode. This will process a request asynchronously (blocking would be disabled).         |
+
+> The Service Extension need to be connected to a deployed [Datadog agent](https://docs.datadoghq.com/agent).
+
+| Environment variable | Default value | Description |
+|---|---|---|
+| `DD_AGENT_HOST` | `N/A` | Host of a running Datadog Agent. |
+| `DD_TRACE_AGENT_PORT` | `8126` | Port of a running Datadog Agent. |
+
+### SSL Configuration
+
+The Envoy of GCP is configured to communicate to the Service Extension with TLS.
+
+`localhost` self signed certificates are generated and bundled into the ASM Service Extension docker image and loaded at the start of the gRPC server.

contrib/haproxy/stream-processing-offload/cmd/spoa/env.go

@@ -0,0 +1,43 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package main
+
+import (
+	"net"
+	"os"
+	"strconv"
+)
+
+// IntEnv returns the parsed int value of an environment variable, or
+// def otherwise.
+func intEnv(key string, def int) int {
+	vv, ok := os.LookupEnv(key)
+	if !ok {
+		return def
+	}
+	v, err := strconv.Atoi(vv)
+	if err != nil {
+		log.Warn("Non-integer value for env var %s, defaulting to %d. Parse failed with error: %v", key, def, err)
+		return def
+	}
+	return v
+}
+
+// IpEnv returns the valid IP value of an environment variable, or def otherwise.
+func ipEnv(key string, def net.IP) net.IP {
+	vv, ok := os.LookupEnv(key)
+	if !ok {
+		return def
+	}
+
+	ip := net.ParseIP(vv)
+	if ip == nil {
+		log.Warn("Non-IP value for env var %s, defaulting to %s", key, def.String())
+		return def
+	}
+
+	return ip
+}

contrib/haproxy/stream-processing-offload/cmd/spoa/main.go

@@ -7,37 +7,155 @@ package main
 
 import (
 	"context"
-	"github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"
+	"errors"
+	"fmt"
 	"net"
+	"net/http"
 	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
+	"time"
 
-	"github.com/DataDog/dd-trace-go/contrib/haproxy/stream-processing-offload/v2"
 	"github.com/negasus/haproxy-spoe-go/agent"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/DataDog/dd-trace-go/contrib/haproxy/stream-processing-offload/v2"
+	"github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"
+	"github.com/DataDog/dd-trace-go/v2/instrumentation"
 )
 
+type haProxySpoaConfig struct {
+	extensionPort        string
+	healthcheckPort      string
+	extensionHost        string
+	bodyParsingSizeLimit int
+}
+
 var log = NewLogger()
 
+func getDefaultEnvVars() map[string]string {
+	return map[string]string{
+		"DD_VERSION":                   instrumentation.Version(), // Version of the tracer
+		"DD_APM_TRACING_ENABLED":       "false",                   // Appsec Standalone
+		"DD_APPSEC_WAF_TIMEOUT":        "10ms",                    // Proxy specific WAF timeout
+		"_DD_APPSEC_PROXY_ENVIRONMENT": "true",                    // Internal config: Enable API Security proxy sampler
+	}
+}
+
+// initializeEnvironment sets up required environment variables with their defaults
+func initializeEnvironment() {
+	for k, v := range getDefaultEnvVars() {
+		if os.Getenv(k) == "" {
+			if err := os.Setenv(k, v); err != nil {
+				log.Error("haproxy_spoa: failed to set %s environment variable: %s\n", k, err.Error())
+			}
+		}
+	}
+}
+
+// loadConfig loads the configuration from the environment variables
+func loadConfig() haProxySpoaConfig {
+	extensionHostStr := ipEnv("DD_HAPROXY_SPOA_HOST", net.IP{0, 0, 0, 0}).String()
+	extensionPortInt := intEnv("DD_HAPROXY_SPOA_PORT", 3000)
+	healthcheckPortInt := intEnv("DD_HAPROXY_SPOA_HEALTHCHECK_PORT", 3080)
+	bodyParsingSizeLimit := intEnv("DD_APPSEC_BODY_PARSING_SIZE_LIMIT", 0)
+
+	extensionPortStr := strconv.FormatInt(int64(extensionPortInt), 10)
+	healthcheckPortStr := strconv.FormatInt(int64(healthcheckPortInt), 10)
+
+	return haProxySpoaConfig{
+		extensionPort:        extensionPortStr,
+		extensionHost:        extensionHostStr,
+		healthcheckPort:      healthcheckPortStr,
+		bodyParsingSizeLimit: bodyParsingSizeLimit,
+	}
+}
+
 func main() {
-	listener, err := net.Listen("tcp4", "127.0.0.1:3000")
-	if err != nil {
-		log.Error("haproxy_spoa: error create listener, %v", err)
+	initializeEnvironment()
+	config := loadConfig()
+
+	if err := startService(config); err != nil {
+		log.Error("haproxy_spoa: %s\n", err.Error())
 		os.Exit(1)
 	}
+
+	log.Info("haproxy_spoa: shutting down\n")
+}
+
+func startService(config haProxySpoaConfig) error {
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+	g, ctx := errgroup.WithContext(ctx)
+
+	g.Go(func() error {
+		return startSpoa(ctx, config)
+	})
+
+	g.Go(func() error {
+		return startHealthCheck(ctx, config)
+	})
+
+	if err := g.Wait(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func startHealthCheck(ctx context.Context, config haProxySpoaConfig) error {
+	muxServer := http.NewServeMux()
+	muxServer.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(`{"status": "ok", "library": {"language": "golang", "version": "` + instrumentation.Version() + `"}}`))
+	})
+
+	server := &http.Server{
+		Addr:    config.extensionHost + ":" + config.healthcheckPort,
+		Handler: muxServer,
+	}
+
+	go func() {
+		<-ctx.Done()
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := server.Shutdown(shutdownCtx); err != nil {
+			log.Error("haproxy_spoa: health check server shutdown: %s\n", err.Error())
+		}
+	}()
+
+	log.Info("haproxy_spoa: health check server started on %s:%s\n", config.extensionHost, config.healthcheckPort)
+	if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return fmt.Errorf("health check http server: %s", err.Error())
+	}
+
+	return nil
+}
+
+func startSpoa(ctx context.Context, config haProxySpoaConfig) error {
+	listener, err := net.Listen("tcp4", config.extensionHost+":"+config.extensionPort)
+	if err != nil {
+		return fmt.Errorf("error creating listener: %w", err)
+	}
 	defer listener.Close()
 
 	_ = tracer.Start(tracer.WithAppSecEnabled(true))
 	defer tracer.Stop()
 
 	appsecHAProxy := streamprocessingoffload.NewHAProxySPOA(streamprocessingoffload.AppsecHAProxyConfig{
 		BlockingUnavailable:  false,
-		BodyParsingSizeLimit: 1000000, // 1MB
-		Context:              context.Background(),
+		BodyParsingSizeLimit: config.bodyParsingSizeLimit,
+		Context:              ctx,
 	})
 
 	a := agent.New(appsecHAProxy.Handler, log)
 
-	log.Info("haproxy_spoa: started\n")
+	log.Info("haproxy_spoa: datadog agent server started on %s:%s\n", config.extensionHost, config.extensionPort)
 	if err := a.Serve(listener); err != nil {
-		log.Error("haproxy_spoa: error agent serve: %+v\n", err)
+		return fmt.Errorf("error starting agent server: %w", err)
 	}
+
+	return nil
 }