diff --git a/aerospike-connection-manager/Dockerfile b/aerospike-connection-manager/Dockerfile index 3d5cdebc..95397c3b 100644 --- a/aerospike-connection-manager/Dockerfile +++ b/aerospike-connection-manager/Dockerfile @@ -1,22 +1,19 @@ -FROM golang:1.20-alpine AS builder +FROM golang:1.23-alpine AS builder ARG VERSION=1.10.0 -ADD . $GOPATH/src/github.com/aerospike/php-client/asld -WORKDIR $GOPATH/src/github.com/aerospike/php-client/asld -RUN go build -ldflags="-X 'main.version=$VERSION'" -o aerospike-connection-manager . \ - && cp aerospike-connection-manager /aerospike-connection-manager +WORKDIR /src +COPY . . +RUN go build -ldflags="-X 'main.version=$VERSION'" -o /aerospike-connection-manager . FROM alpine:latest COPY --from=builder /aerospike-connection-manager /usr/bin/aerospike-connection-manager +COPY asld.toml /etc/aerospike-connection-manager/asld.toml COPY asld.toml.template /etc/aerospike-connection-manager/asld.toml.template -COPY docker-entrypoint.sh /docker-entrypoint.sh - -RUN apk add gettext libintl \ - && chmod +x /docker-entrypoint.sh +# Management HTTP server (metrics, health/readiness) default port. EXPOSE 9145 -ENTRYPOINT [ "/docker-entrypoint.sh" ] -CMD ["aerospike-connection-manager", "--config", "/etc/aerospike-connection-manager/asld.toml"] +ENTRYPOINT ["aerospike-connection-manager"] +CMD ["-config-file", "/etc/aerospike-connection-manager/asld.toml"] diff --git a/aerospike-connection-manager/OBSERVABILITY.md b/aerospike-connection-manager/OBSERVABILITY.md new file mode 100644 index 00000000..50d865b3 --- /dev/null +++ b/aerospike-connection-manager/OBSERVABILITY.md @@ -0,0 +1,213 @@ +# Observability and Operations + +The Aerospike Connection Manager (ACM) exposes operational endpoints — Prometheus +metrics and Kubernetes-style health probes — on a single management HTTP server, +separate from the gRPC data path served over unix sockets. + +- [Management server](#management-server) +- [Configuration](#configuration) + - [Precedence](#precedence) + - [Reference](#reference) + - [Examples](#examples) +- [Endpoints](#endpoints) +- [Metrics](#metrics) +- [Kubernetes integration](#kubernetes-integration) +- [Security](#security) + +## Management server + +When enabled (the default), ACM starts one HTTP server that hosts every +operational endpoint on one port. This follows the common Go and Kubernetes +convention of a dedicated admin port: probes and metric scraping share a single +address while the data plane is untouched. + +The default listen address is `:9145`, matching the `EXPOSE` directive in the +image. Each endpoint can be enabled, disabled and re-pathed independently. + +The server shuts down gracefully on `SIGINT`/`SIGTERM`, draining in-flight +requests before the process exits. + +## Configuration + +Every setting can be provided through three sources, and each is optional. + +### Precedence + +From lowest to highest priority: + +``` +built-in defaults < [management] TOML section < ASLD_* env vars < CLI flags +``` + +The source closest to the process invocation wins: an explicit command-line flag +overrides an environment variable, which overrides the config file, which +overrides the built-in default. A flag only takes effect when it is actually +passed, so it never silently masks a lower-priority source with its default. + +### Reference + +| Setting | TOML (`[management]`) | Environment variable | CLI flag | Default | +|------------------|------------------------------|---------------------------|------------------------|------------------| +| Server enabled | `enabled` | `ASLD_MANAGEMENT_ENABLED` | `-management-enabled` | `true` | +| Listen address | `address` | `ASLD_MANAGEMENT_ADDRESS` | `-management-address` | `:9145` | +| Metrics enabled | `[management.metrics] enabled` | `ASLD_METRICS_ENABLED` | `-metrics-enabled` | `true` | +| Metrics path | `[management.metrics] path` | `ASLD_METRICS_PATH` | `-metrics-path` | `/metrics` | +| Liveness enabled | `[management.liveness] enabled` | `ASLD_LIVENESS_ENABLED` | `-liveness-enabled` | `true` | +| Liveness path | `[management.liveness] path` | `ASLD_LIVENESS_PATH` | `-liveness-path` | `/livez` | +| Readiness enabled| `[management.readiness] enabled` | `ASLD_READINESS_ENABLED` | `-readiness-enabled` | `true` | +| Readiness path | `[management.readiness] path` | `ASLD_READINESS_PATH` | `-readiness-path` | `/readyz` | +| Health enabled | `[management.health] enabled` | `ASLD_HEALTH_ENABLED` | `-health-enabled` | `true` | +| Health path | `[management.health] path` | `ASLD_HEALTH_PATH` | `-health-path` | `/healthz` | +| pprof enabled | `[management.pprof] enabled` | `ASLD_PPROF_ENABLED` | `-pprof-enabled` | `false` | +| pprof path | `[management.pprof] path` | `ASLD_PPROF_PATH` | `-pprof-path` | `/debug/pprof/` | + +The configuration is validated at startup. A non-empty listen address is +required when the server is enabled, every enabled endpoint path must start with +`/`, and two enabled endpoints may not share a path. + +### Examples + +Config file — the `[management]` table lives in the same TOML file as the +cluster definitions: + +```toml +[management] +address = ":9145" + +[management.metrics] +enabled = true +path = "/metrics" + +[management.pprof] +enabled = false +``` + +Environment variables: + +```sh +export ASLD_MANAGEMENT_ADDRESS=":9145" +export ASLD_METRICS_ENABLED=true +export ASLD_PPROF_ENABLED=false +``` + +Command-line flags (override both of the above): + +```sh +aerospike-connection-manager -config-file /etc/aerospike-connection-manager/asld.toml \ + -management-address ":9145" \ + -pprof-enabled=false +``` + +## Endpoints + +| Path | Purpose | +|-----------------|--------------------------------------------------------------------------------------------------| +| `/metrics` | Prometheus / OpenMetrics exposition. | +| `/livez` | Liveness. Cheap, dependency-free. A failure means the process should be restarted. | +| `/readyz` | Readiness. Fails while any configured Aerospike cluster is disconnected. | +| `/healthz` | Aggregate of liveness and readiness, for tooling that expects a single combined endpoint. | +| `/debug/pprof/` | `net/http/pprof` profiling index (opt-in). | + +Liveness and readiness are deliberately distinct. Liveness must not depend on +Aerospike: if it did, a cluster outage would restart otherwise-healthy pods +instead of just draining them. Dependency checks belong on readiness, which +removes the pod from load balancing without a restart. + +The probes return `200` when healthy and `503` otherwise, with a JSON body: + +```json +{ + "status": "error", + "checks": { + "main": "not connected" + } +} +``` + +## Metrics + +All metrics are collected unconditionally; the `/metrics` endpoint only exposes +them when enabled. + +**Go runtime** — `go_goroutines`, `go_threads`, `go_gc_duration_seconds`, +`go_memstats_*`, `go_info`, and related collectors. + +**Process** (on platforms that expose process statistics, e.g. Linux) — +`process_cpu_seconds_total`, `process_resident_memory_bytes`, +`process_open_fds`, `process_max_fds`. + +**gRPC server** — labelled by `grpc_service`, `grpc_method`, `grpc_type` (and +`grpc_code` where applicable): + +- `grpc_server_started_total` +- `grpc_server_handled_total` +- `grpc_server_msg_received_total` +- `grpc_server_msg_sent_total` +- `grpc_server_handling_seconds_*` (histogram; buckets tuned for ACM's + sub-millisecond profile) + +Every method is pre-registered with a zero count, so dashboards are populated +from the first scrape after a deploy rather than only once a method is first +called. + +**Aerospike connection pool** — one series per cluster, labelled `cluster`, +read from the Aerospike client at scrape time: + +| Metric | Type | Meaning | +|-------------------------------------------------|---------|--------------------------------------------------------------------| +| `asld_aerospike_up` | gauge | `1` if the last stats scrape succeeded, `0` otherwise. | +| `asld_aerospike_open_connections` | gauge | Open connections to Aerospike nodes. | +| `asld_aerospike_nodes_total` | gauge | Aerospike nodes currently tracked. | +| `asld_aerospike_connections_attempts_total` | counter | Connection attempts. | +| `asld_aerospike_connections_failed_total` | counter | Failed connections. | +| `asld_aerospike_connections_pool_empty_total` | counter | Times the pool was exhausted and ACM waited or opened a new conn. | +| `asld_aerospike_connections_pool_overflow_total`| counter | Connections dropped because the pool was at capacity. | +| `asld_aerospike_connections_idle_dropped_total` | counter | Connections closed due to idle timeout. | +| `asld_aerospike_tends_failed_total` | counter | Failed cluster-tend cycles (a node could not be reached). | + +A cluster whose stats scrape fails reports `asld_aerospike_up{cluster="…"} 0` +and omits its pool gauges, so one unreachable cluster never hides the others. + +## Kubernetes integration + +Pod spec — expose the management port and wire the probes: + +```yaml +ports: + - name: management + containerPort: 9145 +livenessProbe: + httpGet: + path: /livez + port: management +readinessProbe: + httpGet: + path: /readyz + port: management +``` + +Scraping — with the Prometheus Operator: + +```yaml +apiVersion: monitoring.coreos.com/v1 +kind: PodMonitor +metadata: + name: aerospike-connection-manager +spec: + selector: + matchLabels: + app: aerospike-connection-manager + podMetricsEndpoints: + - port: management + path: /metrics +``` + +## Security + +- `/metrics` and the probes are unauthenticated. Restrict access to the + monitoring network with a `NetworkPolicy` or a service-mesh ACL. +- pprof is disabled by default. It exposes process internals (heap, goroutines, + CPU profiles) without authentication; only enable it for debugging and never + expose it publicly. +- The metric values do not contain request or record contents, but they do + reveal traffic patterns and error rates. diff --git a/aerospike-connection-manager/README.md b/aerospike-connection-manager/README.md index 4b915798..aa8feeed 100644 --- a/aerospike-connection-manager/README.md +++ b/aerospike-connection-manager/README.md @@ -12,6 +12,19 @@ This guide provides step-by-step instructions on setting up the Aerospike Connec ### Configuration Instructions Aerospike's client policy allows for flexible control over read and write operations, including optimistic concurrency, time-to-live settings, and conditional writes based on record existence. The policy may be configured in the existing asld.toml file or you may create a custom toml file. An example asld template toml file is provided below, for reference. +Each Aerospike cluster is declared under a `[clusters.]` table: + +```toml +[clusters.cluster] +socket = "/tmp/asld_grpc.sock" +host = "127.0.0.1:3000" +``` + +Declaring a cluster as a top-level table (e.g. `[cluster]`) is **deprecated** but +still supported; the daemon logs a notice at startup when it reads one, so +existing configs keep working while you migrate. Process-global operational +settings live in the `[management]` section (see [OBSERVABILITY.md](OBSERVABILITY.md)). + 1. **Using the existing asld.toml file to configure the client policy:** - Change directory to php-client/aerospike-connection-manager ```shell @@ -61,6 +74,24 @@ Aerospike's client policy allows for flexible control over read and write operat ```shell sudo make daemonize ``` +  +### Monitoring and Health Checks + +ACM runs a management HTTP server (default `:9145`) that exposes Prometheus +metrics and Kubernetes-style health probes alongside the gRPC data path: + +- `/metrics` — Go runtime, process, gRPC and Aerospike connection-pool metrics +- `/livez` — liveness (cheap, dependency-free) +- `/readyz` — readiness (fails while an Aerospike cluster is disconnected) +- `/healthz` — aggregate of the two +- `/debug/pprof/` — profiling (opt-in, off by default) + +Every endpoint can be enabled, disabled or re-pathed, and the whole server can +be turned off. Settings are resolved with the precedence +`CLI flag > environment variable > config file > default`. See +[OBSERVABILITY.md](OBSERVABILITY.md) for the full configuration reference, +metrics catalogue and Kubernetes examples. +   ### Example asld.toml file: ~~~toml diff --git a/aerospike-connection-manager/aerospike-connection-manager.service b/aerospike-connection-manager/aerospike-connection-manager.service index b9cd35c9..c4a75499 100644 --- a/aerospike-connection-manager/aerospike-connection-manager.service +++ b/aerospike-connection-manager/aerospike-connection-manager.service @@ -5,7 +5,7 @@ Wants=network.target After=network-online.target [Service] -ExecStart=/usr/bin/asld --config-file /etc/aerospike-connection-manager/asld.toml +ExecStart=/usr/bin/asld -config-file /etc/aerospike-connection-manager/asld.toml [Install] WantedBy=multi-user.target \ No newline at end of file diff --git a/aerospike-connection-manager/asld.toml b/aerospike-connection-manager/asld.toml index 3525a743..1ff7d008 100644 --- a/aerospike-connection-manager/asld.toml +++ b/aerospike-connection-manager/asld.toml @@ -4,7 +4,8 @@ # # ----------------------------------------------------- -[cluster] +# Each [clusters.] table defines one Aerospike cluster the daemon serves. +[clusters.cluster] socket = "/tmp/asld_grpc.sock" host = "127.0.0.1:3000" #user = "default-user" @@ -121,3 +122,10 @@ ignore-other-subnet-aliases = false # Peers nodes for the cluster are not discovered and seed nodes are # retained despite connection failures. seed-only-cluster = false + +# Management server (Prometheus metrics and health probes) is enabled by default +# on :9145. Configure it via a [management] section, ASLD_* env vars or CLI +# flags. See OBSERVABILITY.md for the full reference. +# +# [management] +# address = ":9145" diff --git a/aerospike-connection-manager/asld.toml.template b/aerospike-connection-manager/asld.toml.template index 611a0013..274924b2 100644 --- a/aerospike-connection-manager/asld.toml.template +++ b/aerospike-connection-manager/asld.toml.template @@ -2,9 +2,13 @@ # # Aerospike Local Daemon configuration file. # +# Each [clusters.] table defines one Aerospike cluster. Declaring clusters +# as top-level tables (e.g. [cluster]) is deprecated but still supported; the +# daemon logs a notice at startup when it reads one. +# # ----------------------------------------------------- -[cluster] +[clusters.cluster] socket=/tmp/cluster.sock host = "1.1.1.1:3001,2.2.2.2:3002,3.3.3.3" user = "default-user" @@ -122,7 +126,7 @@ ignore-other-subnet-aliases = false # retained despite connection failures. seed-only-cluster = false -[cluster_tls] +[clusters.cluster_tls] port = 4333 host = "3.3.3.3" tls-name = "tls-name" @@ -132,24 +136,24 @@ tls-cafile = "{{.RootCAFile}}" tls-certfile = "{{.CertFile}}" tls-keyfile = "{{.KeyFile}}" -[cluster_instance] +[clusters.cluster_instance] host = "3.3.3.3:3003,4.4.4.4:3004" user = "test-user" password = "test-password" -[cluster_env] +[clusters.cluster_env] host = "5.5.5.5:env-tls-name:1000" password = "env:AEROSPIKE_TEST" -[cluster_envb64] +[clusters.cluster_envb64] host = "6.6.6.6:env-tls-name:1000" password = "env-b64:AEROSPIKE_TEST" -[cluster_b64] +[clusters.cluster_b64] host = "7.7.7.7:env-tls-name:1000" password = "b64:dGVzdC1wYXNzd29yZAo=" -[cluster_file] +[clusters.cluster_file] host = "1.1.1.1" password = "file:{{.PassFile}}" @@ -159,3 +163,44 @@ store-file = "default1.store" [uda_instance] store-file = "test.store" + +# ----------------------------------------------------- +# +# Management server: Prometheus metrics and health probes. +# +# Process-global, not per-cluster. Every value here can also be set via an +# ASLD_* environment variable or a CLI flag; flags override env, env overrides +# this file, this file overrides the built-in defaults. See OBSERVABILITY.md. +# +# The values below are the defaults and can be omitted entirely. +# +# ----------------------------------------------------- + +[management] +# Master switch and listen address for the management HTTP server. +enabled = true +address = ":9145" + +[management.metrics] +enabled = true +path = "/metrics" + +[management.liveness] +# Liveness: cheap, no dependency checks. A failure should restart the process. +enabled = true +path = "/livez" + +[management.readiness] +# Readiness: fails while any Aerospike cluster is disconnected, draining traffic. +enabled = true +path = "/readyz" + +[management.health] +# Aggregate of liveness and readiness. +enabled = true +path = "/healthz" + +[management.pprof] +# net/http/pprof. Disabled by default: it exposes process internals unauthenticated. +enabled = false +path = "/debug/pprof/" diff --git a/aerospike-connection-manager/common/config/conf.go b/aerospike-connection-manager/common/config/conf.go index 5c461cd2..d9d934e0 100644 --- a/aerospike-connection-manager/common/config/conf.go +++ b/aerospike-connection-manager/common/config/conf.go @@ -1,222 +1,159 @@ package config import ( + "fmt" "os" - "time" + "reflect" + "sort" + + "github.com/go-viper/mapstructure/v2" + "github.com/knadh/koanf/providers/confmap" + "github.com/knadh/koanf/v2" + "github.com/pelletier/go-toml/v2" "github.com/aerospike/php-client/asld/common/client" "github.com/aerospike/php-client/asld/common/flags" - "github.com/pelletier/go-toml/v2" ) -func Read(configFile string) (map[string]*client.AerospikeConfig, error) { +const ( + // reservedManagementSection holds process-global operational settings. It is + // parsed separately by internal/config and is never an Aerospike cluster. + reservedManagementSection = "management" + // clustersSection is the table whose sub-tables ([clusters.]) are the + // preferred way to declare Aerospike clusters. + clustersSection = "clusters" +) + +// Read parses the config file and returns the configured Aerospike clusters. +// +// Clusters may be declared under the [clusters.] namespace (preferred) or +// as legacy top-level tables (any table that declares a host or socket). The +// names of the legacy tables are returned separately so the caller can emit a +// deprecation notice; that form is kept only for backward compatibility. +func Read(configFile string) (clusters map[string]*client.AerospikeConfig, legacyClusters []string, err error) { doc, err := os.ReadFile(configFile) if err != nil { - return nil, err + return nil, nil, err } cfg := map[string]map[string]any{} if err := toml.Unmarshal(doc, &cfg); err != nil { - return nil, err + return nil, nil, err } - res := make(map[string]*client.AerospikeConfig, len(cfg)) + clusters = make(map[string]*client.AerospikeConfig, len(cfg)) for section, valMap := range cfg { - f := flags.NewDefaultAerospikeFlags() - - if v, exists := valMap["socket"]; exists { - f.Socket = v.(string) - } - - if v, exists := valMap["host"]; exists { - seeds := flags.NewHostTLSPortSliceFlag() - if err := seeds.Set(v.(string)); err != nil { - return nil, err - } else { - f.Seeds = seeds + switch { + case section == reservedManagementSection: + continue + case section == clustersSection: + if err := decodeNamespace(valMap, clusters); err != nil { + return nil, nil, err } - } - - if v, exists := valMap["port"]; exists { - f.DefaultPort = int(v.(int64)) - } - - if v, exists := valMap["user"]; exists { - f.User = v.(string) - } - - if v, exists := valMap["password"]; exists { - var pass flags.PasswordFlag - if err := pass.Set(v.(string)); err != nil { - return nil, err - } else { - f.Password = pass - } - } - - if v, exists := valMap["auth"]; exists { - var auth flags.AuthModeFlag - if err := auth.Set(v.(string)); err != nil { - return nil, err - } else { - f.AuthMode = auth - } - } - - if v, exists := valMap["tls-enable"]; exists { - f.TLSEnable = v.(bool) - } - - if v, exists := valMap["tls-name"]; exists { - f.TLSName = v.(string) - } - - if v, exists := valMap["tls-protocols"]; exists { - var cv flags.TLSProtocolsFlag - if err := cv.Set(v.(string)); err != nil { - return nil, err - } else { - f.TLSProtocols = cv - } - } - - if v, exists := valMap["tls-cafile"]; exists { - var cv flags.CertFlag - if err := cv.Set(v.(string)); err != nil { - return nil, err - } else { - f.TLSRootCAFile = cv - } - } - - if v, exists := valMap["tls-capath"]; exists { - var cv flags.CertPathFlag - if err := cv.Set(v.(string)); err != nil { - return nil, err - } else { - f.TLSRootCAPath = cv - } - } - - if v, exists := valMap["tls-certfile"]; exists { - var cv flags.CertFlag - if err := cv.Set(v.(string)); err != nil { - return nil, err - } else { - f.TLSCertFile = cv - } - } - - if v, exists := valMap["tls-keyfile"]; exists { - var cv flags.CertFlag - if err := cv.Set(v.(string)); err != nil { - return nil, err - } else { - f.TLSKeyFile = cv + case isClusterSection(valMap): + if err := addCluster(clusters, section, valMap); err != nil { + return nil, nil, err } - } - if v, exists := valMap["tls-keyfile-password"]; exists { - var cv flags.PasswordFlag - if err := cv.Set(v.(string)); err != nil { - return nil, err - } else { - f.TLSKeyFilePass = cv - } + legacyClusters = append(legacyClusters, section) } + } - if v, exists := valMap["cluster-name"]; exists { - f.ClusterName = v.(string) - } + sort.Strings(legacyClusters) - if v, exists := valMap["timeout"]; exists { - v, err := time.ParseDuration(v.(string)) - if err != nil { - return nil, err - } - f.Timeout = v - } + return clusters, legacyClusters, nil +} - if v, exists := valMap["idle-timeout"]; exists { - v, err := time.ParseDuration(v.(string)) - if err != nil { - return nil, err - } - f.IdleTimeout = v +// decodeNamespace decodes every [clusters.] sub-table into res. +func decodeNamespace(valMap map[string]any, res map[string]*client.AerospikeConfig) error { + for name, raw := range valMap { + sub, ok := raw.(map[string]any) + if !ok { + return fmt.Errorf("[clusters.%s] must be a table", name) } - if v, exists := valMap["login-timeout"]; exists { - v, err := time.ParseDuration(v.(string)) - if err != nil { - return nil, err - } - f.LoginTimeout = v + if err := addCluster(res, name, sub); err != nil { + return err } + } - if v, exists := valMap["connection-queue-size"]; exists { - f.ConnectionQueueSize = int(v.(int64)) - } + return nil +} - if v, exists := valMap["min-connections-per-node"]; exists { - f.MinConnectionsPerNode = int(v.(int64)) - } +// addCluster decodes one cluster table and stores it under name, rejecting a +// name used by both the namespaced and the legacy form. +func addCluster(res map[string]*client.AerospikeConfig, name string, valMap map[string]any) error { + if _, dup := res[name]; dup { + return fmt.Errorf("duplicate cluster %q", name) + } - if v, exists := valMap["max-error-rate"]; exists { - f.MaxErrorRate = int(v.(int64)) - } + f := flags.NewDefaultAerospikeFlags() + if err := decodeCluster(valMap, f); err != nil { + return fmt.Errorf("cluster %q: %w", name, err) + } - if v, exists := valMap["error-rate-window"]; exists { - f.ErrorRateWindow = int(v.(int64)) - } + res[name] = f.NewAerospikeConfig() - if v, exists := valMap["limit-connections-to-queue-size"]; exists { - f.LimitConnectionsToQueueSize = v.(bool) - } + return nil +} - if v, exists := valMap["opening-connection-threshold"]; exists { - f.OpeningConnectionThreshold = int(v.(int64)) - } +// isClusterSection reports whether a TOML table defines an Aerospike cluster. +// A cluster must declare where to connect (host) or where to serve (socket); +// tables without either — such as the [uda] agent settings — are not clusters +// and are skipped rather than turned into bogus default-localhost entries. +func isClusterSection(valMap map[string]any) bool { + if _, ok := valMap["host"]; ok { + return true + } - if v, exists := valMap["fail-if-not-connected"]; exists { - f.FailIfNotConnected = v.(bool) - } + _, ok := valMap["socket"] - if v, exists := valMap["tend-interval"]; exists { - v, err := time.ParseDuration(v.(string)) - if err != nil { - return nil, err - } - f.TendInterval = v - } + return ok +} - if v, exists := valMap["use-services-alternate"]; exists { - f.UseServicesAlternate = v.(bool) - } +// decodeCluster maps one TOML cluster table onto the (already toml-tagged) +// AerospikeFlags. The flagValueHook bridges the custom flag types, whose string +// parsing (env/base64/file passwords, host[:tls][:port] seeds, certificate +// loading, auth and TLS protocols) cannot be expressed as plain struct tags. +func decodeCluster(valMap map[string]any, f *flags.AerospikeFlags) error { + k := koanf.New(".") + if err := k.Load(confmap.Provider(valMap, "."), nil); err != nil { + return err + } - if v, exists := valMap["rack-aware"]; exists { - f.RackAware = v.(bool) - } + return k.UnmarshalWithConf("", f, koanf.UnmarshalConf{ + Tag: "toml", + DecoderConfig: &mapstructure.DecoderConfig{ + DecodeHook: mapstructure.ComposeDecodeHookFunc( + flagValueHook, + mapstructure.StringToTimeDurationHookFunc(), + ), + WeaklyTypedInput: true, + Result: f, + }, + }) +} - if v, exists := valMap["rack-ids"]; exists { - v := v.([]any) - res := make([]int, len(v)) - for i := range v { - res[i] = int(v[i].(int64)) - } - f.RackIds = res - } +// flagValueHook decodes a string into any destination type whose pointer +// implements Set(string) error — i.e. the flag.Value types used by +// AerospikeFlags — by delegating to that parser. Other destinations are left +// untouched for the remaining hooks and mapstructure to handle. +func flagValueHook(from, to reflect.Type, data any) (any, error) { + if from.Kind() != reflect.String { + return data, nil + } - if v, exists := valMap["ignore-other-subnet-aliases"]; exists { - f.IgnoreOtherSubnetAliases = v.(bool) - } + ptr := reflect.New(to) - if v, exists := valMap["seed-only-cluster"]; exists { - f.SeedOnlyCluster = v.(bool) - } + setter, ok := ptr.Interface().(interface{ Set(string) error }) + if !ok { + return data, nil + } - res[section] = f.NewAerospikeConfig() + if err := setter.Set(data.(string)); err != nil { + return nil, err } - return res, nil + return ptr.Elem().Interface(), nil } diff --git a/aerospike-connection-manager/common/config/conf_test.go b/aerospike-connection-manager/common/config/conf_test.go new file mode 100644 index 00000000..6c6570ff --- /dev/null +++ b/aerospike-connection-manager/common/config/conf_test.go @@ -0,0 +1,309 @@ +package config + +import ( + "os" + "path/filepath" + "testing" + + as "github.com/aerospike/aerospike-client-go/v7" + + "github.com/aerospike/php-client/asld/common/client" +) + +func writeConfig(t *testing.T, body string) string { + t.Helper() + + path := filepath.Join(t.TempDir(), "asld.toml") + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + + return path +} + +func TestReadSkipsManagementSection(t *testing.T) { + path := writeConfig(t, ` +[cluster] +host = "127.0.0.1:3000" + +[management] +address = ":9145" +metrics-enabled = true +`) + + clusters, _, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + if _, ok := clusters["management"]; ok { + t.Fatal("[management] must not be parsed as an Aerospike cluster") + } + + if _, ok := clusters["cluster"]; !ok { + t.Fatal("real cluster section should be present") + } +} + +func TestReadSkipsNonClusterTables(t *testing.T) { + // [uda]/[uda_instance] declare neither host nor socket, so they must not be + // mistaken for Aerospike clusters. + path := writeConfig(t, ` +[c] +host = "127.0.0.1:3000" + +[uda] +agent-port = 8001 +store-file = "default1.store" + +[uda_instance] +store-file = "test.store" +`) + + clusters, _, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + if len(clusters) != 1 { + t.Fatalf("expected only the real cluster, got %d: %v", len(clusters), keysOf(clusters)) + } + + if _, ok := clusters["c"]; !ok { + t.Fatalf("real cluster missing, got %v", keysOf(clusters)) + } +} + +func TestReadSocketOnlyTableIsCluster(t *testing.T) { + // A table with only a socket (host falls back to the default seed) is still + // a usable cluster and must be kept. + clusters, _, err := Read(writeConfig(t, "[c]\nsocket = \"/tmp/c.sock\"\n")) + if err != nil { + t.Fatalf("Read: %v", err) + } + + if _, ok := clusters["c"]; !ok { + t.Fatalf("socket-only cluster should be kept, got %v", keysOf(clusters)) + } +} + +func keysOf(m map[string]*client.AerospikeConfig) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + + return out +} + +func TestReadScalarsAndDurations(t *testing.T) { + path := writeConfig(t, ` +[c] +socket = "/tmp/c.sock" +host = "127.0.0.1:3000" +user = "alice" +cluster-name = "prod" +timeout = "30s" +idle-timeout = "5s" +connection-queue-size = 100 +max-error-rate = 7 +limit-connections-to-queue-size = true +fail-if-not-connected = true +rack-aware = true +rack-ids = [1, 2, 3] +`) + + ac := mustCluster(t, path) + if ac.Socket != "/tmp/c.sock" { + t.Errorf("socket = %q", ac.Socket) + } + + if ac.User != "alice" { + t.Errorf("user = %q", ac.User) + } + + if ac.ClusterName != "prod" { + t.Errorf("cluster-name = %q", ac.ClusterName) + } + + if ac.Timeout.String() != "30s" || ac.IdleTimeout.String() != "5s" { + t.Errorf("durations = %s / %s", ac.Timeout, ac.IdleTimeout) + } + + if ac.ConnectionQueueSize != 100 || ac.MaxErrorRate != 7 { + t.Errorf("ints = %d / %d", ac.ConnectionQueueSize, ac.MaxErrorRate) + } + + if !ac.LimitConnectionsToQueueSize || !ac.FailIfNotConnected || !ac.RackAware { + t.Errorf("bools not set: %+v", ac) + } + + if len(ac.RackIds) != 3 || ac.RackIds[0] != 1 || ac.RackIds[2] != 3 { + t.Errorf("rack-ids = %v", ac.RackIds) + } +} + +func TestReadPasswordClearAndB64(t *testing.T) { + plain := mustCluster(t, writeConfig(t, "[c]\nhost=\"127.0.0.1\"\npassword=\"s3cret\"\n")) + if plain.Password != "s3cret" { + t.Errorf("clear password = %q", plain.Password) + } + + b64 := mustCluster(t, writeConfig(t, "[c]\nhost=\"127.0.0.1\"\npassword=\"b64:dGVzdA==\"\n")) + if b64.Password != "test" { + t.Errorf("b64 password = %q, want test", b64.Password) + } +} + +func TestReadPasswordFromEnv(t *testing.T) { + t.Setenv("ASLD_TEST_SECRET", "envpass") + + ac := mustCluster(t, writeConfig(t, "[c]\nhost=\"127.0.0.1\"\npassword=\"env:ASLD_TEST_SECRET\"\n")) + if ac.Password != "envpass" { + t.Errorf("env password = %q, want envpass", ac.Password) + } +} + +func TestReadAuthMode(t *testing.T) { + ac := mustCluster(t, writeConfig(t, "[c]\nhost=\"127.0.0.1\"\nauth=\"EXTERNAL\"\n")) + if ac.AuthMode != as.AuthModeExternal { + t.Errorf("auth mode = %v, want EXTERNAL", ac.AuthMode) + } +} + +func TestReadHostParsing(t *testing.T) { + tests := []struct { + name string + host string + wantHost string + wantPort int + wantTLS string + wantLen int + }{ + {"host only, default port", "1.2.3.4", "1.2.3.4", 3000, "", 1}, + {"host and port", "1.2.3.4:5000", "1.2.3.4", 5000, "", 1}, + {"host tls port", "1.2.3.4:tlsname:5001", "1.2.3.4", 5001, "tlsname", 1}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ac := mustCluster(t, writeConfig(t, "[c]\nhost=\""+tc.host+"\"\n")) + if len(ac.Seeds) != tc.wantLen { + t.Fatalf("seeds = %d, want %d", len(ac.Seeds), tc.wantLen) + } + + s := ac.Seeds[0] + if s.Host != tc.wantHost || s.Port != tc.wantPort || s.TLSName != tc.wantTLS { + t.Errorf("seed = %+v, want host=%s port=%d tls=%s", s, tc.wantHost, tc.wantPort, tc.wantTLS) + } + }) + } +} + +func TestReadMultipleSeeds(t *testing.T) { + ac := mustCluster(t, writeConfig(t, "[c]\nhost=\"1.1.1.1:3001,2.2.2.2:3002\"\n")) + if len(ac.Seeds) != 2 { + t.Fatalf("seeds = %d, want 2", len(ac.Seeds)) + } + + if ac.Seeds[0].Port != 3001 || ac.Seeds[1].Port != 3002 { + t.Errorf("ports = %d / %d", ac.Seeds[0].Port, ac.Seeds[1].Port) + } +} + +func TestReadTLSCAFileGatedByEnable(t *testing.T) { + // b64-encoded "ca-bytes" exercises the cert parser without touching the filesystem. + body := func(enable bool) string { + e := "false" + if enable { + e = "true" + } + + return "[c]\nhost=\"127.0.0.1\"\ntls-enable=" + e + "\ntls-cafile=\"b64:Y2EtYnl0ZXM=\"\n" + } + + on := mustCluster(t, writeConfig(t, body(true))) + if len(on.RootCA) != 1 || string(on.RootCA[0]) != "ca-bytes" { + t.Errorf("RootCA with tls-enable = %v", on.RootCA) + } + + off := mustCluster(t, writeConfig(t, body(false))) + if len(off.RootCA) != 0 { + t.Errorf("RootCA must stay empty when tls-enable is false, got %v", off.RootCA) + } +} + +func TestReadClustersNamespace(t *testing.T) { + path := writeConfig(t, ` +[clusters.a] +host = "1.1.1.1:3001" + +[clusters.b] +host = "2.2.2.2:3002" +`) + + clusters, legacy, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + if len(clusters) != 2 || clusters["a"] == nil || clusters["b"] == nil { + t.Fatalf("expected clusters a and b, got %v", keysOf(clusters)) + } + + if len(legacy) != 0 { + t.Errorf("namespaced clusters must not be reported as legacy, got %v", legacy) + } +} + +func TestReadLegacyReportsDeprecation(t *testing.T) { + path := writeConfig(t, ` +[cluster_one] +host = "1.1.1.1:3001" + +[cluster_two] +host = "2.2.2.2:3002" +`) + + clusters, legacy, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + if len(clusters) != 2 { + t.Fatalf("expected 2 clusters, got %v", keysOf(clusters)) + } + + if len(legacy) != 2 || legacy[0] != "cluster_one" || legacy[1] != "cluster_two" { + t.Errorf("legacy clusters = %v, want sorted [cluster_one cluster_two]", legacy) + } +} + +func TestReadDuplicateClusterAcrossForms(t *testing.T) { + path := writeConfig(t, ` +[dup] +host = "1.1.1.1:3001" + +[clusters.dup] +host = "2.2.2.2:3002" +`) + + if _, _, err := Read(path); err == nil { + t.Fatal("expected an error for a cluster name used by both forms") + } +} + +func mustCluster(t *testing.T, path string) *client.AerospikeConfig { + t.Helper() + + clusters, _, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + + ac, ok := clusters["c"] + if !ok { + t.Fatalf("cluster \"c\" not found") + } + + return ac +} diff --git a/aerospike-connection-manager/go.mod b/aerospike-connection-manager/go.mod index 0355d731..f8712e03 100644 --- a/aerospike-connection-manager/go.mod +++ b/aerospike-connection-manager/go.mod @@ -1,22 +1,39 @@ module github.com/aerospike/php-client/asld -go 1.21.3 +go 1.23.0 require ( github.com/aerospike/aerospike-client-go/v7 v7.9.0 - github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1 + github.com/go-viper/mapstructure/v2 v2.4.0 + github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.1.0 + github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 + github.com/knadh/koanf/providers/confmap v1.0.0 + github.com/knadh/koanf/v2 v2.3.5 github.com/pelletier/go-toml/v2 v2.1.1 + github.com/prometheus/client_golang v1.23.2 google.golang.org/grpc v1.63.3 - google.golang.org/protobuf v1.34.2 + google.golang.org/protobuf v1.36.8 ) // replace github.com/aerospike/aerospike-client-go/v7 => /home/khosrow/virt/src/github.com/aerospike/aerospike-client-go require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/knadh/koanf/maps v0.1.2 // indirect + github.com/kr/text v0.2.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect + github.com/mitchellh/copystructure v1.2.0 // indirect + github.com/mitchellh/reflectwalk v1.0.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect github.com/yuin/gopher-lua v1.1.1 // indirect - golang.org/x/net v0.26.0 // indirect - golang.org/x/sync v0.7.0 // indirect - golang.org/x/sys v0.21.0 // indirect - golang.org/x/text v0.16.0 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect + golang.org/x/net v0.43.0 // indirect + golang.org/x/sync v0.16.0 // indirect + golang.org/x/sys v0.35.0 // indirect + golang.org/x/text v0.28.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d // indirect ) diff --git a/aerospike-connection-manager/go.sum b/aerospike-connection-manager/go.sum index 21a27509..c2f776f7 100644 --- a/aerospike-connection-manager/go.sum +++ b/aerospike-connection-manager/go.sum @@ -1,5 +1,10 @@ github.com/aerospike/aerospike-client-go/v7 v7.9.0 h1:RFuJIirAa+jVM859y55KTiMsbVHWRDKedKxsm8HX3Ok= github.com/aerospike/aerospike-client-go/v7 v7.9.0/go.mod h1:STlBtOkKT8nmp7iD+sEkr/JGEOu+4e2jGlNN0Jiu2a4= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -7,14 +12,38 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI= github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls= +github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= +github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= -github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= -github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/pprof v0.0.0-20240711041743-f6c9dda6c6da h1:xRmpO92tb8y+Z85iUOMOicpCfaYcv7o3Cg3wKrIpg8g= github.com/google/pprof v0.0.0-20240711041743-f6c9dda6c6da/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo= -github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1 h1:HcUWd006luQPljE73d5sk+/VgYPGUReEVz2y1/qylwY= -github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.1/go.mod h1:w9Y7gY31krpLmrVU5ZPG9H7l9fZuRu5/3R3S3FMtVQ4= +github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.1.0 h1:QGLs/O40yoNK9vmy4rhUGBVyMf1lISBGtXRpsu/Qu/o= +github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.1.0/go.mod h1:hM2alZsMUni80N33RBe6J0e423LB+odMj7d3EMP9l20= +github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 h1:pRhl55Yx1eC7BZ1N+BBWwnKaMyD8uC+34TLdndZMAKk= +github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0/go.mod h1:XKMd7iuf/RGPSMJ/U4HP0zS2Z9Fh8Ps9a+6X26m/tmI= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo= +github.com/knadh/koanf/maps v0.1.2/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI= +github.com/knadh/koanf/providers/confmap v1.0.0 h1:mHKLJTE7iXEys6deO5p6olAiZdG5zwp8Aebir+/EaRE= +github.com/knadh/koanf/providers/confmap v1.0.0/go.mod h1:txHYHiI2hAtF0/0sCmcuol4IDcuQbKTybiB1nOcUo1A= +github.com/knadh/koanf/v2 v2.3.5 h1:2dXJUYaKGm4SGYeoAtBviq9+02JZo/pxQ2ssOd60rJg= +github.com/knadh/koanf/v2 v2.3.5/go.mod h1:gRb40VRAbd4iJMYYD5IxZ6hfuopFcXBpc9bbQpZwo28= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw= +github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s= +github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ= +github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/onsi/ginkgo/v2 v2.16.0 h1:7q1w9frJDzninhXxjZd+Y/x54XNjG/UlRLIYPZafsPM= github.com/onsi/ginkgo/v2 v2.16.0/go.mod h1:llBI3WDLL9Z6taip6f33H76YcWtJv+7R3HigUjbIBOs= github.com/onsi/gomega v1.32.0 h1:JRYU78fJ1LPxlckP6Txi/EYqJvjtMrDC04/MM5XRHPk= @@ -23,32 +52,49 @@ github.com/pelletier/go-toml/v2 v2.1.1 h1:LWAJwfNvjQZCFIDKWYQaM62NcYeYViCmWIwmOS github.com/pelletier/go-toml/v2 v2.1.1/go.mod h1:tJU2Z3ZkXwnxa4DPO899bsyIoywizdUvyaeZurnPPDc= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= -golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ= -golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE= -golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= -golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= -golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws= -golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4= -golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI= -golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg= -golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= +golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= +golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= +golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw= +golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= +golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= +golang.org/x/tools v0.35.0 h1:mBffYraMEf7aa0sB+NuKnuCy8qI/9Bughn8dC2Gu5r0= +golang.org/x/tools v0.35.0/go.mod h1:NKdj5HkL/73byiZSJjqJgKn3ep7KjFkBOkR/Hps3VPw= google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d h1:JU0iKnSg02Gmb5ZdV8nYsKEKsP6o/FGVWTrw4i1DA9A= google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d/go.mod h1:Ue6ibwXGpU+dqIcODieyLOcgj7z8+IcskoNIgZxtrFY= google.golang.org/grpc v1.63.3 h1:FGVegD7MHo/zhaGduk/R85WvSFJ+si70UQIJ0fg+BiU= google.golang.org/grpc v1.63.3/go.mod h1:5FFeE/YiGPD2flWFCrCx8K3Ay7hALATnKiI8U3avIuw= -google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= -google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/aerospike-connection-manager/internal/config/load.go b/aerospike-connection-manager/internal/config/load.go new file mode 100644 index 00000000..e99ee24a --- /dev/null +++ b/aerospike-connection-manager/internal/config/load.go @@ -0,0 +1,49 @@ +package config + +import ( + "os" + + "github.com/pelletier/go-toml/v2" +) + +// FileSection is the reserved TOML table name that holds management settings. +// common/config skips this section so it is not mistaken for an Aerospike +// cluster definition. +const FileSection = "management" + +// readFileSection reads the [management] table from the TOML config file. A +// missing file or a missing section is not an error: it simply means no +// file-level overrides are present and the defaults (plus env and flags) apply. +func readFileSection(configFile string) (map[string]any, error) { + doc, err := os.ReadFile(configFile) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + + cfg := map[string]map[string]any{} + if err := toml.Unmarshal(doc, &cfg); err != nil { + return nil, err + } + return cfg[FileSection], nil +} + +// Load resolves the management configuration from the config file, environment +// variables and CLI flags, in that ascending order of priority. collectFlags +// is the closure returned by RegisterFlags after flag.Parse has run; pass nil +// when no flags are wired (the file and environment are still applied). +func Load(configFile string, collectFlags func() map[string]string) (Management, error) { + section, err := readFileSection(configFile) + if err != nil { + return Management{}, err + } + + var flags map[string]string + if collectFlags != nil { + flags = collectFlags() + } + + return Resolve(section, FromEnv(os.LookupEnv), flags) +} diff --git a/aerospike-connection-manager/internal/config/load_test.go b/aerospike-connection-manager/internal/config/load_test.go new file mode 100644 index 00000000..bfa1355a --- /dev/null +++ b/aerospike-connection-manager/internal/config/load_test.go @@ -0,0 +1,67 @@ +package config + +import ( + "os" + "path/filepath" + "testing" +) + +func writeTempConfig(t *testing.T, body string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "asld.toml") + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatalf("write temp config: %v", err) + } + return path +} + +func TestLoadFromFile(t *testing.T) { + path := writeTempConfig(t, ` +[cluster] +host = "127.0.0.1:3000" + +[management] +address = ":7777" + +[management.pprof] +enabled = true + +[management.metrics] +path = "/custom-metrics" +`) + + m, err := Load(path, nil) + if err != nil { + t.Fatalf("Load: %v", err) + } + if m.Address != ":7777" { + t.Errorf("address = %q, want :7777", m.Address) + } + if !m.Pprof.Enabled { + t.Error("pprof should be enabled from file") + } + if m.Metrics.Path != "/custom-metrics" { + t.Errorf("metrics path = %q", m.Metrics.Path) + } +} + +func TestLoadMissingFile(t *testing.T) { + m, err := Load(filepath.Join(t.TempDir(), "does-not-exist.toml"), nil) + if err != nil { + t.Fatalf("missing file should not error: %v", err) + } + if m.Address != DefaultAddress { + t.Errorf("address = %q, want default", m.Address) + } +} + +func TestLoadMissingManagementSection(t *testing.T) { + path := writeTempConfig(t, "[cluster]\nhost = \"127.0.0.1:3000\"\n") + m, err := Load(path, nil) + if err != nil { + t.Fatalf("Load: %v", err) + } + if m != Default() { + t.Errorf("config without [management] should equal Default(), got %+v", m) + } +} diff --git a/aerospike-connection-manager/internal/config/management.go b/aerospike-connection-manager/internal/config/management.go new file mode 100644 index 00000000..3e17a0f1 --- /dev/null +++ b/aerospike-connection-manager/internal/config/management.go @@ -0,0 +1,316 @@ +// Package config resolves the operational ("management") configuration of the +// Aerospike Connection Manager from layered sources. +// +// The management configuration controls the HTTP admin server that exposes +// Prometheus metrics and Kubernetes-style health probes. It is intentionally +// separate from the per-cluster Aerospike configuration handled by +// common/config, because these settings are process-global rather than +// per-cluster. +// +// Resolution precedence, from lowest to highest, is: +// +// built-in defaults < [management] TOML section < ASLD_* env vars < CLI flags +// +// The closer a source is to the invocation of the process, the higher its +// priority: an explicit command-line flag always wins, an environment variable +// overrides the config file, and the config file overrides the defaults. +// +// Source merging and struct mapping are delegated to knadh/koanf; this package +// only declares the field schema and the conventional env/flag names. +package config + +import ( + "flag" + "fmt" + "strings" + + "github.com/go-viper/mapstructure/v2" + "github.com/knadh/koanf/providers/confmap" + "github.com/knadh/koanf/v2" +) + +// DefaultAddress is the listen address of the management HTTP server when no +// other source overrides it. It matches the EXPOSE directive in the Dockerfile. +const DefaultAddress = ":9145" + +// keyDelim separates nested keys in the koanf store and in the logical keys +// produced by the source adapters (e.g. "metrics.enabled"). +const keyDelim = "." + +// Endpoint describes a single HTTP endpoint served by the management server. +type Endpoint struct { + Enabled bool `koanf:"enabled"` + // Path is the URL the endpoint is served on; it must start with "/". + Path string `koanf:"path"` +} + +// Management is the fully resolved operational configuration. +type Management struct { + // Enabled is the master switch for the management HTTP server. When false, + // no admin server is started regardless of the individual endpoint toggles. + Enabled bool `koanf:"enabled"` + // Address is the host:port the management HTTP server listens on. + Address string `koanf:"address"` + + // Metrics serves Prometheus metrics (Go runtime, process, gRPC, Aerospike). + Metrics Endpoint `koanf:"metrics"` + // Liveness answers "is the process alive" — cheap, no dependency checks. + Liveness Endpoint `koanf:"liveness"` + // Readiness answers "can the process serve traffic" — checks Aerospike + // connectivity for every configured cluster. + Readiness Endpoint `koanf:"readiness"` + // Health is the aggregate of liveness and readiness checks. + Health Endpoint `koanf:"health"` + // Pprof serves net/http/pprof debug endpoints. Disabled by default because + // it exposes process internals without authentication. + Pprof Endpoint `koanf:"pprof"` +} + +// Default returns the built-in management configuration used as the base of the +// resolution chain. +func Default() Management { + return Management{ + Enabled: true, + Address: DefaultAddress, + Metrics: Endpoint{Enabled: true, Path: "/metrics"}, + Liveness: Endpoint{Enabled: true, Path: "/livez"}, + Readiness: Endpoint{Enabled: true, Path: "/readyz"}, + Health: Endpoint{Enabled: true, Path: "/healthz"}, + Pprof: Endpoint{Enabled: false, Path: "/debug/pprof/"}, + } +} + +// Logical keys identify a single configurable field across all sources. They +// double as koanf paths, so nesting them with keyDelim maps straight onto the +// struct via the koanf tags above. +const ( + keyEnabled = "enabled" + keyAddress = "address" + keyMetricsEnabled = "metrics.enabled" + keyMetricsPath = "metrics.path" + keyLivenessEnabled = "liveness.enabled" + keyLivenessPath = "liveness.path" + keyReadinessEnabled = "readiness.enabled" + keyReadinessPath = "readiness.path" + keyHealthEnabled = "health.enabled" + keyHealthPath = "health.path" + keyPprofEnabled = "pprof.enabled" + keyPprofPath = "pprof.path" +) + +// envKeys maps logical keys to their ASLD_* environment variable names. +var envKeys = map[string]string{ + keyEnabled: "ASLD_MANAGEMENT_ENABLED", + keyAddress: "ASLD_MANAGEMENT_ADDRESS", + keyMetricsEnabled: "ASLD_METRICS_ENABLED", + keyMetricsPath: "ASLD_METRICS_PATH", + keyLivenessEnabled: "ASLD_LIVENESS_ENABLED", + keyLivenessPath: "ASLD_LIVENESS_PATH", + keyReadinessEnabled: "ASLD_READINESS_ENABLED", + keyReadinessPath: "ASLD_READINESS_PATH", + keyHealthEnabled: "ASLD_HEALTH_ENABLED", + keyHealthPath: "ASLD_HEALTH_PATH", + keyPprofEnabled: "ASLD_PPROF_ENABLED", + keyPprofPath: "ASLD_PPROF_PATH", +} + +// flagKeys maps logical keys to their CLI flag names. +var flagKeys = map[string]string{ + keyEnabled: "management-enabled", + keyAddress: "management-address", + keyMetricsEnabled: "metrics-enabled", + keyMetricsPath: "metrics-path", + keyLivenessEnabled: "liveness-enabled", + keyLivenessPath: "liveness-path", + keyReadinessEnabled: "readiness-enabled", + keyReadinessPath: "readiness-path", + keyHealthEnabled: "health-enabled", + keyHealthPath: "health-path", + keyPprofEnabled: "pprof-enabled", + keyPprofPath: "pprof-path", +} + +// boolKeys is the set of logical keys whose value is a boolean. Used to choose +// the flag type during registration. +var boolKeys = map[string]bool{ + keyEnabled: true, + keyMetricsEnabled: true, + keyLivenessEnabled: true, + keyReadinessEnabled: true, + keyHealthEnabled: true, + keyPprofEnabled: true, +} + +// Resolve overlays the file, environment and flag sources onto the built-in +// defaults and returns the validated configuration. The file source is the +// parsed [management] TOML table (nil if absent); env and flag sources are +// logical-key maps produced by FromEnv and the RegisterFlags collector. Each +// source overrides the previous one. +func Resolve(file map[string]any, env, flags map[string]string) (Management, error) { + k := koanf.New(keyDelim) + + if len(file) > 0 { + if err := k.Load(confmap.Provider(file, keyDelim), nil); err != nil { + return Management{}, fmt.Errorf("load file config: %w", err) + } + } + if err := k.Load(confmap.Provider(toAny(env), keyDelim), nil); err != nil { + return Management{}, fmt.Errorf("load env config: %w", err) + } + if err := k.Load(confmap.Provider(toAny(flags), keyDelim), nil); err != nil { + return Management{}, fmt.Errorf("load flag config: %w", err) + } + + // Unmarshal onto the defaults so any field absent from every source keeps + // its built-in value. WeaklyTypedInput lets the string-valued env and flag + // sources decode into bool fields ("true" -> true). + m := Default() + if err := k.UnmarshalWithConf("", &m, koanf.UnmarshalConf{ + Tag: "koanf", + DecoderConfig: &mapstructure.DecoderConfig{ + WeaklyTypedInput: true, + Result: &m, + }, + }); err != nil { + return Management{}, fmt.Errorf("decode config: %w", err) + } + + if err := m.validate(); err != nil { + return Management{}, err + } + return m, nil +} + +// toAny widens a string-keyed map for confmap, which works with any-valued +// maps. The dotted keys are un-nested by koanf using keyDelim. +func toAny(in map[string]string) map[string]any { + out := make(map[string]any, len(in)) + for k, v := range in { + out[k] = v + } + return out +} + +// validate rejects configurations that would fail at runtime, e.g. an empty +// listen address or two enabled endpoints sharing the same path (which would +// panic the HTTP mux on registration). +func (m Management) validate() error { + if !m.Enabled { + return nil + } + if strings.TrimSpace(m.Address) == "" { + return fmt.Errorf("management address must not be empty when the management server is enabled") + } + + endpoints := []struct { + name string + ep Endpoint + }{ + {"metrics", m.Metrics}, + {"liveness", m.Liveness}, + {"readiness", m.Readiness}, + {"health", m.Health}, + {"pprof", m.Pprof}, + } + + seen := make(map[string]string, len(endpoints)) + for _, e := range endpoints { + if !e.ep.Enabled { + continue + } + if !strings.HasPrefix(e.ep.Path, "/") { + return fmt.Errorf("%s path %q must start with %q", e.name, e.ep.Path, "/") + } + if other, dup := seen[e.ep.Path]; dup { + return fmt.Errorf("%s and %s endpoints share the same path %q", other, e.name, e.ep.Path) + } + seen[e.ep.Path] = e.name + } + return nil +} + +// FromEnv extracts management settings from environment variables using the +// supplied lookup function (typically os.LookupEnv). Only variables that are +// actually set contribute to the result. +func FromEnv(lookup func(string) (string, bool)) map[string]string { + out := make(map[string]string) + for logical, env := range envKeys { + if v, ok := lookup(env); ok { + out[logical] = v + } + } + return out +} + +// RegisterFlags defines the management CLI flags on the given flag set and +// returns a collector that, once flags are parsed, yields only the flags that +// were explicitly set. Relying on explicit-set semantics (via FlagSet.Visit) +// is what lets a flag override env and file only when the user actually passes +// it, leaving lower-priority sources intact otherwise. +func RegisterFlags(fs *flag.FlagSet) func() map[string]string { + def := Default() + for logical, name := range flagKeys { + if boolKeys[logical] { + fs.Bool(name, defaultBool(def, logical), usage(logical)) + } else { + fs.String(name, defaultString(def, logical), usage(logical)) + } + } + + flagToLogical := make(map[string]string, len(flagKeys)) + for logical, name := range flagKeys { + flagToLogical[name] = logical + } + + return func() map[string]string { + out := make(map[string]string) + fs.Visit(func(f *flag.Flag) { + if logical, ok := flagToLogical[f.Name]; ok { + out[logical] = f.Value.String() + } + }) + return out + } +} + +func usage(logical string) string { + return fmt.Sprintf("management setting %q (overrides %s and the config file)", logical, envKeys[logical]) +} + +func defaultBool(m Management, logical string) bool { + switch logical { + case keyEnabled: + return m.Enabled + case keyMetricsEnabled: + return m.Metrics.Enabled + case keyLivenessEnabled: + return m.Liveness.Enabled + case keyReadinessEnabled: + return m.Readiness.Enabled + case keyHealthEnabled: + return m.Health.Enabled + case keyPprofEnabled: + return m.Pprof.Enabled + default: + return false + } +} + +func defaultString(m Management, logical string) string { + switch logical { + case keyAddress: + return m.Address + case keyMetricsPath: + return m.Metrics.Path + case keyLivenessPath: + return m.Liveness.Path + case keyReadinessPath: + return m.Readiness.Path + case keyHealthPath: + return m.Health.Path + case keyPprofPath: + return m.Pprof.Path + default: + return "" + } +} diff --git a/aerospike-connection-manager/internal/config/management_test.go b/aerospike-connection-manager/internal/config/management_test.go new file mode 100644 index 00000000..8cd21fed --- /dev/null +++ b/aerospike-connection-manager/internal/config/management_test.go @@ -0,0 +1,159 @@ +package config + +import ( + "flag" + "testing" +) + +func TestDefault(t *testing.T) { + d := Default() + if !d.Enabled { + t.Fatal("management server should be enabled by default") + } + if d.Address != DefaultAddress { + t.Fatalf("default address = %q, want %q", d.Address, DefaultAddress) + } + if d.Pprof.Enabled { + t.Fatal("pprof must be disabled by default for security") + } + if d.Liveness.Path != "/livez" || d.Readiness.Path != "/readyz" || d.Health.Path != "/healthz" { + t.Fatalf("unexpected default probe paths: %+v", d) + } +} + +func TestResolvePrecedence(t *testing.T) { + file := map[string]any{ + "address": ":1111", + "metrics": map[string]any{"path": "/file-metrics", "enabled": true}, + } + env := map[string]string{ + keyAddress: ":2222", + keyMetricsPath: "/env-metrics", + } + flags := map[string]string{ + keyAddress: ":3333", + } + + m, err := Resolve(file, env, flags) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + + if m.Address != ":3333" { + t.Errorf("address = %q, want flag value :3333", m.Address) + } + if m.Metrics.Path != "/env-metrics" { + t.Errorf("metrics path = %q, want env value /env-metrics", m.Metrics.Path) + } + if !m.Metrics.Enabled { + t.Error("metrics enabled should remain true from file") + } + if m.Readiness.Path != "/readyz" { + t.Errorf("readiness path = %q, want default /readyz", m.Readiness.Path) + } +} + +func TestResolveBoolOverride(t *testing.T) { + m, err := Resolve(nil, map[string]string{keyPprofEnabled: "true"}, map[string]string{keyMetricsEnabled: "false"}) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if !m.Pprof.Enabled { + t.Error("pprof should be enabled via env") + } + if m.Metrics.Enabled { + t.Error("metrics should be disabled via flag") + } +} + +func TestResolveInvalidBool(t *testing.T) { + if _, err := Resolve(nil, map[string]string{keyEnabled: "notabool"}, nil); err == nil { + t.Fatal("expected error for invalid boolean") + } +} + +func TestValidate(t *testing.T) { + tests := []struct { + name string + env map[string]string + wantErr bool + }{ + {"empty address", map[string]string{keyAddress: " "}, true}, + {"path without slash", map[string]string{keyMetricsPath: "metrics"}, true}, + {"duplicate paths", map[string]string{keyLivenessPath: "/x", keyReadinessPath: "/x"}, true}, + {"disabled duplicate is fine", map[string]string{keyLivenessPath: "/x", keyReadinessPath: "/x", keyReadinessEnabled: "false"}, false}, + {"disabled server skips validation", map[string]string{keyEnabled: "false", keyAddress: ""}, false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, err := Resolve(nil, tc.env, nil) + if (err != nil) != tc.wantErr { + t.Fatalf("Resolve err = %v, wantErr = %v", err, tc.wantErr) + } + }) + } +} + +func TestResolveFileSource(t *testing.T) { + file := map[string]any{ + "address": ":7000", + "metrics": map[string]any{"enabled": false}, + "readiness": map[string]any{"path": "/ready"}, + } + m, err := Resolve(file, nil, nil) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if m.Address != ":7000" || m.Metrics.Enabled || m.Readiness.Path != "/ready" { + t.Fatalf("file source not applied: %+v", m) + } +} + +func TestFromEnv(t *testing.T) { + values := map[string]string{ + "ASLD_MANAGEMENT_ADDRESS": ":9999", + "ASLD_PPROF_ENABLED": "true", + } + lookup := func(k string) (string, bool) { + v, ok := values[k] + return v, ok + } + src := FromEnv(lookup) + if src[keyAddress] != ":9999" { + t.Errorf("address = %q", src[keyAddress]) + } + if src[keyPprofEnabled] != "true" { + t.Errorf("pprof enabled = %q", src[keyPprofEnabled]) + } + if _, ok := src[keyMetricsPath]; ok { + t.Error("unset env var must not appear in source map") + } +} + +func TestRegisterFlagsExplicitOnly(t *testing.T) { + fs := flag.NewFlagSet("test", flag.ContinueOnError) + collect := RegisterFlags(fs) + + if err := fs.Parse([]string{"-management-address=:8888", "-pprof-enabled=true"}); err != nil { + t.Fatalf("parse: %v", err) + } + + src := collect() + if len(src) != 2 { + t.Fatalf("expected only the 2 explicitly-set flags, got %+v", src) + } + if src[keyAddress] != ":8888" || src[keyPprofEnabled] != "true" { + t.Fatalf("unexpected collected flags: %+v", src) + } + + m, err := Resolve(nil, nil, src) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + if m.Address != ":8888" || !m.Pprof.Enabled { + t.Fatalf("flags not applied: %+v", m) + } + if m.Metrics.Path != "/metrics" { + t.Errorf("metrics path = %q, want default", m.Metrics.Path) + } +} diff --git a/aerospike-connection-manager/internal/health/health.go b/aerospike-connection-manager/internal/health/health.go new file mode 100644 index 00000000..c10c2c04 --- /dev/null +++ b/aerospike-connection-manager/internal/health/health.go @@ -0,0 +1,166 @@ +// Package health provides Kubernetes-style liveness and readiness probes. +// +// The distinction follows standard practice and matters operationally: +// +// - Liveness ("/livez") answers "is the process still working?". It must be +// cheap and must not depend on external systems — if it did, a transient +// dependency outage would make Kubernetes restart otherwise-healthy pods. +// - Readiness ("/readyz") answers "can the process serve traffic right now?". +// This is where dependency checks belong (Aerospike connectivity): when it +// fails the pod is removed from load balancing but not restarted. +// - Health ("/healthz") is the aggregate of both, kept for compatibility with +// tooling that expects the legacy combined endpoint. +package health + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sort" + "sync" + "time" +) + +// defaultTimeout bounds how long the whole set of checks for one request may +// run, so a single hung check cannot wedge a probe indefinitely. +const defaultTimeout = 2 * time.Second + +// Check reports the health of a single subsystem. A nil error means healthy. +type Check func(ctx context.Context) error + +// Connectivity is anything that can report whether it is currently connected, +// such as the Aerospike client. It keeps this package free of an Aerospike +// dependency while still offering a ready-made readiness check. +type Connectivity interface { + IsConnected() bool +} + +// Connected adapts a Connectivity into a Check that fails when the dependency +// reports itself as disconnected. +func Connected(c Connectivity) Check { + return func(context.Context) error { + if !c.IsConnected() { + return fmt.Errorf("not connected") + } + return nil + } +} + +type namedCheck struct { + name string + check Check +} + +// Checker aggregates liveness and readiness checks and exposes them as HTTP +// handlers. The zero value is not usable; call NewChecker. +type Checker struct { + // Timeout bounds the execution of all checks for a single request. + Timeout time.Duration + + mu sync.RWMutex + liveness []namedCheck + readiness []namedCheck +} + +// NewChecker returns an empty Checker with the default per-request timeout. +func NewChecker() *Checker { + return &Checker{Timeout: defaultTimeout} +} + +// AddLiveness registers a liveness check. Liveness checks should be cheap and +// free of external dependencies. +func (c *Checker) AddLiveness(name string, check Check) { + c.mu.Lock() + defer c.mu.Unlock() + c.liveness = append(c.liveness, namedCheck{name, check}) +} + +// AddReadiness registers a readiness check, typically a dependency probe. +func (c *Checker) AddReadiness(name string, check Check) { + c.mu.Lock() + defer c.mu.Unlock() + c.readiness = append(c.readiness, namedCheck{name, check}) +} + +// LivenessHandler serves the liveness probe. +func (c *Checker) LivenessHandler() http.Handler { + return c.handler(func() []namedCheck { return c.snapshot(true, false) }) +} + +// ReadinessHandler serves the readiness probe. +func (c *Checker) ReadinessHandler() http.Handler { + return c.handler(func() []namedCheck { return c.snapshot(false, true) }) +} + +// HealthHandler serves the aggregate of liveness and readiness checks. +func (c *Checker) HealthHandler() http.Handler { + return c.handler(func() []namedCheck { return c.snapshot(true, true) }) +} + +// snapshot returns a copy of the requested check sets so checks run without +// holding the lock. +func (c *Checker) snapshot(live, ready bool) []namedCheck { + c.mu.RLock() + defer c.mu.RUnlock() + out := make([]namedCheck, 0, len(c.liveness)+len(c.readiness)) + if live { + out = append(out, c.liveness...) + } + if ready { + out = append(out, c.readiness...) + } + return out +} + +type response struct { + Status string `json:"status"` + Checks map[string]string `json:"checks,omitempty"` +} + +func (c *Checker) handler(selector func() []namedCheck) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + timeout := c.Timeout + if timeout <= 0 { + timeout = defaultTimeout + } + ctx, cancel := context.WithTimeout(r.Context(), timeout) + defer cancel() + + checks := selector() + results := make(map[string]string, len(checks)) + healthy := true + for _, nc := range checks { + if err := nc.check(ctx); err != nil { + results[nc.name] = err.Error() + healthy = false + } else { + results[nc.name] = "ok" + } + } + + body := response{Status: "ok", Checks: results} + code := http.StatusOK + if !healthy { + body.Status = "error" + code = http.StatusServiceUnavailable + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(body) + } +} + +// CheckNames returns the registered readiness check names, sorted. Intended for +// diagnostics and tests. +func (c *Checker) CheckNames() []string { + c.mu.RLock() + defer c.mu.RUnlock() + names := make([]string, 0, len(c.readiness)) + for _, nc := range c.readiness { + names = append(names, nc.name) + } + sort.Strings(names) + return names +} diff --git a/aerospike-connection-manager/internal/health/health_test.go b/aerospike-connection-manager/internal/health/health_test.go new file mode 100644 index 00000000..46b4309a --- /dev/null +++ b/aerospike-connection-manager/internal/health/health_test.go @@ -0,0 +1,107 @@ +package health + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" +) + +type fakeConn struct{ connected bool } + +func (f fakeConn) IsConnected() bool { return f.connected } + +func do(t *testing.T, h http.Handler) (int, response) { + t.Helper() + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil)) + var body response + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body %q: %v", rec.Body.String(), err) + } + return rec.Code, body +} + +func TestLivenessNoChecks(t *testing.T) { + c := NewChecker() + code, body := do(t, c.LivenessHandler()) + if code != http.StatusOK || body.Status != "ok" { + t.Fatalf("liveness = %d %q, want 200 ok", code, body.Status) + } +} + +func TestReadinessPasses(t *testing.T) { + c := NewChecker() + c.AddReadiness("aerospike", Connected(fakeConn{connected: true})) + + code, body := do(t, c.ReadinessHandler()) + if code != http.StatusOK { + t.Fatalf("status = %d, want 200", code) + } + if body.Checks["aerospike"] != "ok" { + t.Fatalf("check result = %q, want ok", body.Checks["aerospike"]) + } +} + +func TestReadinessFails(t *testing.T) { + c := NewChecker() + c.AddReadiness("aerospike", Connected(fakeConn{connected: false})) + + code, body := do(t, c.ReadinessHandler()) + if code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want 503", code) + } + if body.Status != "error" { + t.Fatalf("status field = %q, want error", body.Status) + } + if body.Checks["aerospike"] != "not connected" { + t.Fatalf("check detail = %q", body.Checks["aerospike"]) + } +} + +func TestLivenessIndependentOfReadiness(t *testing.T) { + c := NewChecker() + c.AddReadiness("aerospike", Connected(fakeConn{connected: false})) + + // A failing dependency must not affect liveness — otherwise an Aerospike + // outage would trigger pod restarts instead of just removing from routing. + if code, _ := do(t, c.LivenessHandler()); code != http.StatusOK { + t.Fatalf("liveness status = %d, want 200 despite failing readiness", code) + } +} + +func TestHealthAggregates(t *testing.T) { + c := NewChecker() + c.AddLiveness("self", func(context.Context) error { return nil }) + c.AddReadiness("aerospike", Connected(fakeConn{connected: false})) + + code, body := do(t, c.HealthHandler()) + if code != http.StatusServiceUnavailable { + t.Fatalf("healthz status = %d, want 503", code) + } + if body.Checks["self"] != "ok" || body.Checks["aerospike"] == "ok" { + t.Fatalf("healthz should report both checks: %+v", body.Checks) + } +} + +func TestReadinessReportsCheckError(t *testing.T) { + c := NewChecker() + c.AddReadiness("custom", func(context.Context) error { return errors.New("boom") }) + + code, body := do(t, c.ReadinessHandler()) + if code != http.StatusServiceUnavailable || body.Checks["custom"] != "boom" { + t.Fatalf("got %d %+v", code, body.Checks) + } +} + +func TestCheckNames(t *testing.T) { + c := NewChecker() + c.AddReadiness("b", Connected(fakeConn{connected: true})) + c.AddReadiness("a", Connected(fakeConn{connected: true})) + names := c.CheckNames() + if len(names) != 2 || names[0] != "a" || names[1] != "b" { + t.Fatalf("CheckNames = %v, want [a b]", names) + } +} diff --git a/aerospike-connection-manager/internal/management/server.go b/aerospike-connection-manager/internal/management/server.go new file mode 100644 index 00000000..105cdf1a --- /dev/null +++ b/aerospike-connection-manager/internal/management/server.go @@ -0,0 +1,130 @@ +// Package management hosts the operational HTTP server of the Aerospike +// Connection Manager. +// +// It serves Prometheus metrics and the liveness/readiness/health probes on a +// single admin port, separate from the gRPC data path. Following the common Go +// and Kubernetes convention of one admin port keeps probe configuration and +// metric scraping simple while leaving the data plane untouched. Each route is +// mounted only when enabled in the resolved configuration. +package management + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "net/http/pprof" + "path" + "time" + + "github.com/aerospike/php-client/asld/internal/config" + "github.com/aerospike/php-client/asld/internal/health" +) + +// readHeaderTimeout guards the admin server against slow-loris style stalls. It +// is generous because the endpoints are internal and not latency sensitive. +const readHeaderTimeout = 5 * time.Second + +// Server is the management HTTP server. Use New to construct it. +type Server struct { + cfg config.Management + http *http.Server + logger *slog.Logger + + listener net.Listener +} + +// New assembles the management server from the resolved configuration. The +// metrics handler may be nil when metrics are disabled; the checker may be nil +// when no health endpoints are enabled. When the management server is disabled +// the returned Server is inert and Start/Shutdown are no-ops. +func New(cfg config.Management, metricsHandler http.Handler, checker *health.Checker, logger *slog.Logger) *Server { + if logger == nil { + logger = slog.Default() + } + + mux := http.NewServeMux() + + if cfg.Metrics.Enabled && metricsHandler != nil { + mux.Handle(cfg.Metrics.Path, metricsHandler) + } + if checker != nil { + if cfg.Liveness.Enabled { + mux.Handle(cfg.Liveness.Path, checker.LivenessHandler()) + } + if cfg.Readiness.Enabled { + mux.Handle(cfg.Readiness.Path, checker.ReadinessHandler()) + } + if cfg.Health.Enabled { + mux.Handle(cfg.Health.Path, checker.HealthHandler()) + } + } + if cfg.Pprof.Enabled { + registerPprof(mux, cfg.Pprof.Path) + } + + return &Server{ + cfg: cfg, + logger: logger, + http: &http.Server{ + Addr: cfg.Address, + Handler: mux, + ReadHeaderTimeout: readHeaderTimeout, + }, + } +} + +// registerPprof mounts the standard net/http/pprof handlers under prefix, +// mirroring how the package registers itself on the default mux. +func registerPprof(mux *http.ServeMux, prefix string) { + mux.HandleFunc(prefix, pprof.Index) + mux.HandleFunc(path.Join(prefix, "cmdline"), pprof.Cmdline) + mux.HandleFunc(path.Join(prefix, "profile"), pprof.Profile) + mux.HandleFunc(path.Join(prefix, "symbol"), pprof.Symbol) + mux.HandleFunc(path.Join(prefix, "trace"), pprof.Trace) +} + +// Start binds the listen address and serves in the background. The bind happens +// synchronously so address-in-use errors surface to the caller at startup +// rather than being lost in a goroutine. It is a no-op when the management +// server is disabled. +func (s *Server) Start() error { + if !s.cfg.Enabled { + s.logger.Info("management server disabled") + return nil + } + + ln, err := net.Listen("tcp", s.http.Addr) + if err != nil { + return fmt.Errorf("management listen on %s: %w", s.http.Addr, err) + } + s.listener = ln + + s.logger.Info("management server listening", "address", ln.Addr().String()) + go func() { + if err := s.http.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + s.logger.Error("management server stopped unexpectedly", "err", err) + } + }() + return nil +} + +// Shutdown gracefully stops the management server, waiting for in-flight +// requests until ctx is done. It is a no-op when the server was never started. +func (s *Server) Shutdown(ctx context.Context) error { + if !s.cfg.Enabled || s.listener == nil { + return nil + } + return s.http.Shutdown(ctx) +} + +// Addr returns the actual listen address, which is useful when the configured +// address used port 0. It returns an empty string before Start. +func (s *Server) Addr() string { + if s.listener == nil { + return "" + } + return s.listener.Addr().String() +} diff --git a/aerospike-connection-manager/internal/management/server_test.go b/aerospike-connection-manager/internal/management/server_test.go new file mode 100644 index 00000000..ac3d25f1 --- /dev/null +++ b/aerospike-connection-manager/internal/management/server_test.go @@ -0,0 +1,141 @@ +package management + +import ( + "context" + "net/http" + "testing" + "time" + + "github.com/aerospike/php-client/asld/internal/config" + "github.com/aerospike/php-client/asld/internal/health" +) + +func stubMetrics() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("# metrics")) + }) +} + +func allEnabled() config.Management { + c := config.Default() + c.Address = "127.0.0.1:0" + c.Pprof.Enabled = true + return c +} + +func startServer(t *testing.T, cfg config.Management) *Server { + t.Helper() + checker := health.NewChecker() + checker.AddReadiness("aerospike", health.Connected(connFake{true})) + + s := New(cfg, stubMetrics(), checker, nil) + if err := s.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = s.Shutdown(ctx) + }) + return s +} + +type connFake struct{ ok bool } + +func (c connFake) IsConnected() bool { return c.ok } + +func get(t *testing.T, url string) int { + t.Helper() + resp, err := http.Get(url) + if err != nil { + t.Fatalf("GET %s: %v", url, err) + } + defer resp.Body.Close() + return resp.StatusCode +} + +func TestServerRoutesEnabled(t *testing.T) { + s := startServer(t, allEnabled()) + base := "http://" + s.Addr() + + for path, want := range map[string]int{ + "/metrics": http.StatusOK, + "/livez": http.StatusOK, + "/readyz": http.StatusOK, + "/healthz": http.StatusOK, + "/debug/pprof/": http.StatusOK, + "/debug/pprof/cmdline": http.StatusOK, + } { + if code := get(t, base+path); code != want { + t.Errorf("GET %s = %d, want %d", path, code, want) + } + } +} + +func TestServerDisabledEndpointReturns404(t *testing.T) { + cfg := allEnabled() + cfg.Metrics.Enabled = false + s := startServer(t, cfg) + + if code := get(t, "http://"+s.Addr()+"/metrics"); code != http.StatusNotFound { + t.Errorf("disabled /metrics = %d, want 404", code) + } + // Probes remain available. + if code := get(t, "http://"+s.Addr()+"/readyz"); code != http.StatusOK { + t.Errorf("/readyz = %d, want 200", code) + } +} + +func TestServerReadinessReflectsDependency(t *testing.T) { + checker := health.NewChecker() + checker.AddReadiness("aerospike", health.Connected(connFake{false})) + + cfg := allEnabled() + s := New(cfg, stubMetrics(), checker, nil) + if err := s.Start(); err != nil { + t.Fatalf("Start: %v", err) + } + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = s.Shutdown(ctx) + }() + + if code := get(t, "http://"+s.Addr()+"/readyz"); code != http.StatusServiceUnavailable { + t.Errorf("/readyz with disconnected dependency = %d, want 503", code) + } +} + +func TestServerDisabledIsInert(t *testing.T) { + cfg := config.Default() + cfg.Enabled = false + s := New(cfg, stubMetrics(), health.NewChecker(), nil) + if err := s.Start(); err != nil { + t.Fatalf("Start on disabled server: %v", err) + } + if s.Addr() != "" { + t.Errorf("disabled server should not bind, got %q", s.Addr()) + } + if err := s.Shutdown(context.Background()); err != nil { + t.Errorf("Shutdown on disabled server: %v", err) + } +} + +func TestServerStartBindError(t *testing.T) { + cfg := config.Default() + cfg.Address = "127.0.0.1:0" + first := New(cfg, stubMetrics(), health.NewChecker(), nil) + if err := first.Start(); err != nil { + t.Fatalf("first Start: %v", err) + } + defer func() { _ = first.Shutdown(context.Background()) }() + + // Re-binding the already-taken address must surface synchronously. + clash := config.Default() + clash.Address = first.Addr() + second := New(clash, stubMetrics(), health.NewChecker(), nil) + if err := second.Start(); err == nil { + _ = second.Shutdown(context.Background()) + t.Fatal("expected bind error on duplicate address") + } +} diff --git a/aerospike-connection-manager/internal/metrics/aerospike.go b/aerospike-connection-manager/internal/metrics/aerospike.go new file mode 100644 index 00000000..2ba5b05c --- /dev/null +++ b/aerospike-connection-manager/internal/metrics/aerospike.go @@ -0,0 +1,140 @@ +package metrics + +import ( + "sync" + + "github.com/prometheus/client_golang/prometheus" +) + +const ( + aerospikeNamespace = "asld" + aerospikeSubsystem = "aerospike" +) + +// StatsProvider is the subset of the Aerospike client used to read +// connection-pool statistics. Keeping it as a local interface (returning the +// standard error type) leaves this package free of an Aerospike dependency and +// lets tests drive the collector with a fake; the main package adapts +// *aerospike.Client onto it. +type StatsProvider interface { + Stats() (map[string]interface{}, error) +} + +// aerospikeCollector is a prometheus.Collector that reads connection-pool +// statistics from one or more Aerospike clients at scrape time. Reading on +// scrape (rather than polling on a timer) keeps the values fresh, removes a +// background goroutine, and lets Prometheus drive the collection cadence. +type aerospikeCollector struct { + mu sync.RWMutex + providers map[string]StatsProvider + + up *prometheus.Desc + openConnections *prometheus.Desc + nodesTotal *prometheus.Desc + connAttempts *prometheus.Desc + connFailed *prometheus.Desc + poolEmpty *prometheus.Desc + poolOverflow *prometheus.Desc + idleDropped *prometheus.Desc + tendsFailed *prometheus.Desc +} + +func newAerospikeCollector() *aerospikeCollector { + labels := []string{"cluster"} + desc := func(name, help string) *prometheus.Desc { + return prometheus.NewDesc( + prometheus.BuildFQName(aerospikeNamespace, aerospikeSubsystem, name), + help, labels, nil, + ) + } + return &aerospikeCollector{ + providers: make(map[string]StatsProvider), + up: desc("up", "Whether the last Aerospike stats scrape for the cluster succeeded (1) or failed (0)."), + openConnections: desc("open_connections", "Current number of open connections from asld to Aerospike nodes."), + nodesTotal: desc("nodes_total", "Number of Aerospike nodes currently tracked by the client."), + connAttempts: desc("connections_attempts_total", "Cumulative connection attempts from asld to Aerospike."), + connFailed: desc("connections_failed_total", "Cumulative failed connections from asld to Aerospike."), + poolEmpty: desc("connections_pool_empty_total", "Cumulative times the connection pool was exhausted and asld had to wait or open a new connection."), + poolOverflow: desc("connections_pool_overflow_total", "Cumulative connections dropped because the pool was already at capacity."), + idleDropped: desc("connections_idle_dropped_total", "Cumulative connections closed because they exceeded the idle timeout."), + tendsFailed: desc("tends_failed_total", "Cumulative failed cluster-tend cycles (asld could not reach an Aerospike node)."), + } +} + +// register adds a cluster's stats provider to the collector. Safe to call +// concurrently with scrapes. +func (c *aerospikeCollector) register(cluster string, p StatsProvider) { + c.mu.Lock() + defer c.mu.Unlock() + c.providers[cluster] = p +} + +// Describe implements prometheus.Collector. +func (c *aerospikeCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- c.up + ch <- c.openConnections + ch <- c.nodesTotal + ch <- c.connAttempts + ch <- c.connFailed + ch <- c.poolEmpty + ch <- c.poolOverflow + ch <- c.idleDropped + ch <- c.tendsFailed +} + +// Collect implements prometheus.Collector. A cluster whose Stats call fails is +// reported with up=0 and otherwise omitted, so a single unreachable cluster +// never blocks the metrics of the others. +func (c *aerospikeCollector) Collect(ch chan<- prometheus.Metric) { + c.mu.RLock() + providers := make(map[string]StatsProvider, len(c.providers)) + for name, p := range c.providers { + providers[name] = p + } + c.mu.RUnlock() + + for cluster, provider := range providers { + stats, err := provider.Stats() + if err != nil || stats == nil { + ch <- prometheus.MustNewConstMetric(c.up, prometheus.GaugeValue, 0, cluster) + continue + } + ch <- prometheus.MustNewConstMetric(c.up, prometheus.GaugeValue, 1, cluster) + + ch <- prometheus.MustNewConstMetric(c.openConnections, prometheus.GaugeValue, statFloat(stats, "open-connections"), cluster) + ch <- prometheus.MustNewConstMetric(c.nodesTotal, prometheus.GaugeValue, statFloat(stats, "total-nodes"), cluster) + + agg := aggregatedStats(stats) + ch <- prometheus.MustNewConstMetric(c.connAttempts, prometheus.CounterValue, statFloat(agg, "connections-attempts"), cluster) + ch <- prometheus.MustNewConstMetric(c.connFailed, prometheus.CounterValue, statFloat(agg, "connections-failed"), cluster) + ch <- prometheus.MustNewConstMetric(c.poolEmpty, prometheus.CounterValue, statFloat(agg, "connections-pool-empty"), cluster) + ch <- prometheus.MustNewConstMetric(c.poolOverflow, prometheus.CounterValue, statFloat(agg, "connections-pool-overflow"), cluster) + ch <- prometheus.MustNewConstMetric(c.idleDropped, prometheus.CounterValue, statFloat(agg, "connections-idle-dropped"), cluster) + ch <- prometheus.MustNewConstMetric(c.tendsFailed, prometheus.CounterValue, statFloat(agg, "tends-failed"), cluster) + } +} + +// aggregatedStats returns the cluster-aggregated-stats sub-map produced by the +// Aerospike client, or an empty map when it is absent. +func aggregatedStats(stats map[string]interface{}) map[string]interface{} { + if agg, ok := stats["cluster-aggregated-stats"].(map[string]interface{}); ok { + return agg + } + return map[string]interface{}{} +} + +// statFloat reads a numeric stat regardless of whether it arrived as a native +// int (top-level values set by the client) or a float64 (everything that has +// been JSON round-tripped through the client's Stats implementation). +func statFloat(m map[string]interface{}, key string) float64 { + switch v := m[key].(type) { + case float64: + return v + case int: + return float64(v) + case int64: + return float64(v) + default: + return 0 + } +} diff --git a/aerospike-connection-manager/internal/metrics/aerospike_test.go b/aerospike-connection-manager/internal/metrics/aerospike_test.go new file mode 100644 index 00000000..34dd261d --- /dev/null +++ b/aerospike-connection-manager/internal/metrics/aerospike_test.go @@ -0,0 +1,98 @@ +package metrics + +import ( + "errors" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" +) + +type fakeStats struct { + stats map[string]interface{} + err error +} + +func (f fakeStats) Stats() (map[string]interface{}, error) { + return f.stats, f.err +} + +func healthyStats() map[string]interface{} { + return map[string]interface{}{ + "open-connections": 12, + "total-nodes": 3, + "cluster-aggregated-stats": map[string]interface{}{ + "connections-attempts": float64(100), + "connections-failed": float64(2), + "connections-pool-empty": float64(5), + "connections-pool-overflow": float64(1), + "connections-idle-dropped": float64(7), + "tends-failed": float64(0), + }, + } +} + +func TestAerospikeCollectorHealthy(t *testing.T) { + m := New() + m.RegisterAerospike("main", fakeStats{stats: healthyStats()}) + + expected := ` +# HELP asld_aerospike_open_connections Current number of open connections from asld to Aerospike nodes. +# TYPE asld_aerospike_open_connections gauge +asld_aerospike_open_connections{cluster="main"} 12 +# HELP asld_aerospike_up Whether the last Aerospike stats scrape for the cluster succeeded (1) or failed (0). +# TYPE asld_aerospike_up gauge +asld_aerospike_up{cluster="main"} 1 +` + if err := testutil.GatherAndCompare(m.Registry(), strings.NewReader(expected), + "asld_aerospike_open_connections", "asld_aerospike_up"); err != nil { + t.Fatal(err) + } +} + +func TestAerospikeCollectorCounters(t *testing.T) { + m := New() + m.RegisterAerospike("main", fakeStats{stats: healthyStats()}) + + expected := ` +# HELP asld_aerospike_connections_failed_total Cumulative failed connections from asld to Aerospike. +# TYPE asld_aerospike_connections_failed_total counter +asld_aerospike_connections_failed_total{cluster="main"} 2 +# HELP asld_aerospike_tends_failed_total Cumulative failed cluster-tend cycles (asld could not reach an Aerospike node). +# TYPE asld_aerospike_tends_failed_total counter +asld_aerospike_tends_failed_total{cluster="main"} 0 +` + if err := testutil.GatherAndCompare(m.Registry(), strings.NewReader(expected), + "asld_aerospike_connections_failed_total", "asld_aerospike_tends_failed_total"); err != nil { + t.Fatal(err) + } +} + +func TestAerospikeCollectorScrapeError(t *testing.T) { + m := New() + m.RegisterAerospike("down", fakeStats{err: errors.New("not connected")}) + + expected := ` +# HELP asld_aerospike_up Whether the last Aerospike stats scrape for the cluster succeeded (1) or failed (0). +# TYPE asld_aerospike_up gauge +asld_aerospike_up{cluster="down"} 0 +` + if err := testutil.GatherAndCompare(m.Registry(), strings.NewReader(expected), "asld_aerospike_up"); err != nil { + t.Fatal(err) + } + + // A failed scrape must not emit the pool gauges for that cluster. + if n := testutil.CollectAndCount(m.Registry(), "asld_aerospike_open_connections"); n != 0 { + t.Errorf("open_connections series count = %d, want 0 for a failed scrape", n) + } +} + +func TestAerospikeCollectorMultiCluster(t *testing.T) { + m := New() + m.RegisterAerospike("a", fakeStats{stats: healthyStats()}) + m.RegisterAerospike("b", fakeStats{stats: healthyStats()}) + + if n := testutil.CollectAndCount(m.Registry(), "asld_aerospike_up"); n != 2 { + t.Errorf("up series count = %d, want 2 (one per cluster)", n) + } +} diff --git a/aerospike-connection-manager/internal/metrics/metrics.go b/aerospike-connection-manager/internal/metrics/metrics.go new file mode 100644 index 00000000..df6ba554 --- /dev/null +++ b/aerospike-connection-manager/internal/metrics/metrics.go @@ -0,0 +1,111 @@ +// Package metrics owns the Prometheus instrumentation of the Aerospike +// Connection Manager. +// +// It bundles three layers of metrics behind a single registry: +// +// - Go runtime metrics (goroutines, GC, memory) via the standard Go collector. +// - Process metrics (CPU, resident memory, open file descriptors) where the +// platform exposes them. +// - gRPC server metrics (per-method request counts and a latency histogram) +// via a server interceptor. +// +// Per-cluster Aerospike connection-pool metrics are added by RegisterAerospike. +// +// The instrumentation is always collected; whether it is exposed is decided by +// the management server, which only mounts Handler when the metrics endpoint is +// enabled. +package metrics + +import ( + "net/http" + + grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" + "github.com/prometheus/client_golang/prometheus/promhttp" + "google.golang.org/grpc" +) + +// grpcLatencyBuckets are tuned for the asld gRPC profile: sub-millisecond +// median, low-millisecond tail. The default Prometheus buckets start at 5ms and +// would collapse almost every Aerospike call into a single bucket, making the +// histogram useless for latency analysis. +var grpcLatencyBuckets = []float64{ + 0.0001, 0.00025, 0.0005, + 0.001, 0.0025, 0.005, + 0.01, 0.025, 0.05, + 0.1, 0.25, 0.5, 1.0, +} + +// Metrics owns the Prometheus registry and the instrumentation registered on +// it. A single instance is shared across all gRPC servers in the process. +type Metrics struct { + registry *prometheus.Registry + grpcMetrics *grpcprom.ServerMetrics + aerospike *aerospikeCollector +} + +// New constructs a Metrics instance with the Go runtime, process and gRPC +// server collectors pre-registered. +func New() *Metrics { + registry := prometheus.NewRegistry() + registry.MustRegister( + collectors.NewGoCollector(), + collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), + ) + + grpcMetrics := grpcprom.NewServerMetrics( + grpcprom.WithServerHandlingTimeHistogram( + grpcprom.WithHistogramBuckets(grpcLatencyBuckets), + ), + ) + registry.MustRegister(grpcMetrics) + + aerospike := newAerospikeCollector() + registry.MustRegister(aerospike) + + return &Metrics{ + registry: registry, + grpcMetrics: grpcMetrics, + aerospike: aerospike, + } +} + +// Registry exposes the underlying Prometheus registry, mainly for testing. +func (m *Metrics) Registry() *prometheus.Registry { + return m.registry +} + +// UnaryServerInterceptor returns the gRPC unary interceptor that records +// per-method request counts and latencies. +func (m *Metrics) UnaryServerInterceptor() grpc.UnaryServerInterceptor { + return m.grpcMetrics.UnaryServerInterceptor() +} + +// StreamServerInterceptor returns the gRPC stream interceptor counterpart. +func (m *Metrics) StreamServerInterceptor() grpc.StreamServerInterceptor { + return m.grpcMetrics.StreamServerInterceptor() +} + +// InitializeServer pre-registers every method of srv with a zero count so that +// dashboards are populated immediately after deploy rather than only once each +// method has been exercised for the first time. +func (m *Metrics) InitializeServer(srv *grpc.Server) { + m.grpcMetrics.InitializeMetrics(srv) +} + +// RegisterAerospike adds a cluster's client to the connection-pool collector. +// Stats are read lazily on each scrape, so this only needs to be called once +// per cluster before the metrics endpoint starts serving. +func (m *Metrics) RegisterAerospike(cluster string, provider StatsProvider) { + m.aerospike.register(cluster, provider) +} + +// Handler returns the HTTP handler that serves the registry in the Prometheus +// text and OpenMetrics formats. +func (m *Metrics) Handler() http.Handler { + return promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{ + EnableOpenMetrics: true, + Registry: m.registry, + }) +} diff --git a/aerospike-connection-manager/internal/metrics/metrics_test.go b/aerospike-connection-manager/internal/metrics/metrics_test.go new file mode 100644 index 00000000..4f9b49e0 --- /dev/null +++ b/aerospike-connection-manager/internal/metrics/metrics_test.go @@ -0,0 +1,70 @@ +package metrics + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestNewRegistersRuntimeCollectors(t *testing.T) { + m := New() + + families, err := m.Registry().Gather() + if err != nil { + t.Fatalf("Gather: %v", err) + } + if len(families) == 0 { + t.Fatal("expected at least the Go runtime collector to register metrics") + } + + names := make(map[string]bool, len(families)) + for _, f := range families { + names[f.GetName()] = true + } + if !names["go_goroutines"] { + t.Errorf("go_goroutines not exposed; got families %v", keys(names)) + } +} + +func TestHandlerServesPrometheusFormat(t *testing.T) { + m := New() + srv := httptest.NewServer(m.Handler()) + defer srv.Close() + + resp, err := http.Get(srv.URL) + if err != nil { + t.Fatalf("GET: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if !strings.Contains(string(body), "go_goroutines") { + t.Errorf("metrics output missing go_goroutines:\n%s", body) + } +} + +func TestInterceptorsNonNil(t *testing.T) { + m := New() + if m.UnaryServerInterceptor() == nil { + t.Error("unary interceptor is nil") + } + if m.StreamServerInterceptor() == nil { + t.Error("stream interceptor is nil") + } +} + +func keys(m map[string]bool) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} diff --git a/aerospike-connection-manager/main.go b/aerospike-connection-manager/main.go index 10603455..44fe24e5 100644 --- a/aerospike-connection-manager/main.go +++ b/aerospike-connection-manager/main.go @@ -1,9 +1,12 @@ package main import ( + "context" + "errors" "flag" + "fmt" "io/fs" - "log" + "log/slog" "net" "os" "os/signal" @@ -14,7 +17,7 @@ import ( "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery" "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/health" + grpchealth "google.golang.org/grpc/health" "google.golang.org/grpc/health/grpc_health_v1" "google.golang.org/grpc/reflection" "google.golang.org/grpc/status" @@ -23,138 +26,219 @@ import ( "github.com/aerospike/php-client/asld/common/client" "github.com/aerospike/php-client/asld/common/config" + mgmtconfig "github.com/aerospike/php-client/asld/internal/config" + "github.com/aerospike/php-client/asld/internal/health" + "github.com/aerospike/php-client/asld/internal/management" + "github.com/aerospike/php-client/asld/internal/metrics" pb "github.com/aerospike/php-client/asld/proto" ) var ( - version = "0.1.0" - revision = "N/A" - lastCommit time.Time + version = "0.1.0" + revision = "N/A" ) -// TODO: Finish logging and make sure logs have prefixes for different clusters +// shutdownTimeout bounds the graceful shutdown of the management server before +// the process forces its way down. +const shutdownTimeout = 15 * time.Second + +// defaultConnectionQueueSize is applied when the cluster policy leaves the +// connection queue unset, matching the historical asld default. +const defaultConnectionQueueSize = 32 + +// clusterServer bundles everything needed to serve and later tear down a single +// Aerospike cluster's gRPC endpoint. +type clusterServer struct { + name string + grpc *grpc.Server + listener net.Listener + client *aero.Client +} func main() { - var ( - configFile = flag.String("config-file", "/etc/aerospike-connection-manager/asld.toml", "Config File") - showUsage = flag.Bool("h", false, "Show usage information") - showVersion = flag.Bool("v", false, "Print version") - ) + configFile := flag.String("config-file", "/etc/aerospike-connection-manager/asld.toml", "Config File") + showUsage := flag.Bool("h", false, "Show usage information") + showVersion := flag.Bool("v", false, "Print version") + collectManagementFlags := mgmtconfig.RegisterFlags(flag.CommandLine) flag.Parse() if *showUsage { flag.Usage() os.Exit(0) } - if *showVersion { - println(version) + fmt.Println(version) os.Exit(0) } - log.Printf("Aerospike Local Proxy `%s`.", version) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})) + slog.SetDefault(logger) + logger.Info("starting Aerospike Connection Manager", "version", version, "revision", revision) + + clusters, legacyClusters, err := config.Read(*configFile) + if err != nil { + logger.Error("failed to read config", "file", *configFile, "err", err) + os.Exit(1) + } + if len(clusters) == 0 { + logger.Error("no Aerospike clusters defined in config", "file", *configFile) + os.Exit(1) + } + if len(legacyClusters) > 0 { + logger.Warn("deprecated cluster configuration: declare clusters under [clusters.]; "+ + "top-level cluster tables are deprecated and support will be removed in a future release", + "legacy_clusters", legacyClusters) + } - conf, err := config.Read(*configFile) + managementCfg, err := mgmtconfig.Load(*configFile, collectManagementFlags) if err != nil { - log.Fatalln(err) + logger.Error("failed to resolve management config", "err", err) + os.Exit(1) } - defer cleanUp(conf) - c := make(chan os.Signal, 1) - signal.Notify(c, os.Interrupt, syscall.SIGTERM) - go func() { - <-c - cleanUp(conf) + m := metrics.New() + checker := health.NewChecker() + + servers := make([]*clusterServer, 0, len(clusters)) + for name, ac := range clusters { + cs, err := setupCluster(name, ac, m, logger) + if err != nil { + logger.Error("failed to set up cluster", "cluster", name, "err", err) + shutdownServers(servers) + cleanUp(clusters, logger) + os.Exit(1) + } + checker.AddReadiness(name, health.Connected(cs.client)) + m.RegisterAerospike(name, aerospikeStats{cs.client}) + servers = append(servers, cs) + } + + managementSrv := management.New(managementCfg, m.Handler(), checker, logger) + if err := managementSrv.Start(); err != nil { + logger.Error("failed to start management server", "err", err) + shutdownServers(servers) + cleanUp(clusters, logger) os.Exit(1) - }() + } - for cluster, ac := range conf { - go launchServer(cluster, ac) + serveErr := make(chan error, len(servers)) + for _, cs := range servers { + cs := cs + go func() { + logger.Info("serving cluster", "cluster", cs.name, "socket", cs.listener.Addr().String()) + if err := cs.grpc.Serve(cs.listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { + serveErr <- fmt.Errorf("cluster %q: %w", cs.name, err) + } + }() } - e := make(chan struct{}, 1) - <-e -} + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() -func cleanUp(conf map[string]*client.AerospikeConfig) { - for _, ac := range conf { - log.Printf("cleaning up socket `%s`.", ac.Socket) - err := os.Remove(ac.Socket) - if err != nil && err != os.ErrNotExist && err != fs.ErrNotExist { - log.Printf("Socket %s was not cleaned up: %s.", ac.Socket, err) - } + select { + case <-ctx.Done(): + logger.Info("shutdown signal received, draining") + case err := <-serveErr: + logger.Error("gRPC server failed, shutting down", "err", err) } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) + defer cancel() + if err := managementSrv.Shutdown(shutdownCtx); err != nil { + logger.Error("management server shutdown error", "err", err) + } + shutdownServers(servers) + cleanUp(clusters, logger) + logger.Info("shutdown complete") } -func launchServer(name string, ac *client.AerospikeConfig) { +// setupCluster builds, but does not start, the gRPC server for a single +// cluster. The Aerospike client, the unix socket listener and the gRPC server +// (with metrics and panic-recovery interceptors) are all created here so that +// any failure is reported to the caller before serving begins. +func setupCluster(name string, ac *client.AerospikeConfig, m *metrics.Metrics, logger *slog.Logger) (*clusterServer, error) { cp, err := ac.NewClientPolicy() if err != nil { - log.Fatalln(err) + return nil, fmt.Errorf("client policy: %w", err) } - if cp.ConnectionQueueSize == 0 { - cp.ConnectionQueueSize = 32 + cp.ConnectionQueueSize = defaultConnectionQueueSize } - seeds := ac.NewHosts() - - client, err := aero.NewClientWithPolicyAndHost(cp, seeds...) - if err != nil { - log.Fatalln(err) + c, aerr := aero.NewClientWithPolicyAndHost(cp, ac.NewHosts()...) + if aerr != nil { + return nil, fmt.Errorf("connect: %w", aerr) + } + if _, werr := c.WarmUp(-1); werr != nil { + logger.Warn("connection pool warm-up incomplete", "cluster", name, "err", werr) } - client.WarmUp(-1) - log.Printf("Server is Initializing for cluster `%s`. There will be cake...", name) ln, err := net.Listen("unix", ac.Socket) if err != nil { - log.Printf("Server initialization failed: %s", err) - log.Fatalln("The cake was a lie!") + c.Close() + return nil, fmt.Errorf("listen on socket %s: %w", ac.Socket, err) } - defer os.Remove(ac.Socket) - - // tcpLn, err := net.Listen(PROTOCOL_TCP, ADDR) - // if err != nil { - // log.Fatal(err) - // } - - grpcPanicRecoveryHandler := func(p any) (err error) { - log.Println("recovered from panic", "panic", p, "stack", string(debug.Stack())) + recoveryHandler := func(p any) error { + logger.Error("recovered from panic", "cluster", name, "panic", p, "stack", string(debug.Stack())) return status.Errorf(codes.Internal, "%s", p) } srv := grpc.NewServer( - // set the maximum message size possible for a record: 128MiB for memory namespaces, with overhead + // Allow the largest record possible: 128MiB for memory namespaces, with overhead. grpc.MaxRecvMsgSize(130*1024*1024), grpc.MaxSendMsgSize(130*1024*1024), - grpc.ChainUnaryInterceptor(recovery.UnaryServerInterceptor(recovery.WithRecoveryHandler(grpcPanicRecoveryHandler))), - grpc.ChainStreamInterceptor(recovery.StreamServerInterceptor(recovery.WithRecoveryHandler(grpcPanicRecoveryHandler))), + grpc.ChainUnaryInterceptor( + m.UnaryServerInterceptor(), + recovery.UnaryServerInterceptor(recovery.WithRecoveryHandler(recoveryHandler)), + ), + grpc.ChainStreamInterceptor( + m.StreamServerInterceptor(), + recovery.StreamServerInterceptor(recovery.WithRecoveryHandler(recoveryHandler)), + ), ) - grpc_health_v1.RegisterHealthServer(srv, health.NewServer()) - pb.RegisterKVSServer(srv, &server{client: client}) + grpc_health_v1.RegisterHealthServer(srv, grpchealth.NewServer()) + pb.RegisterKVSServer(srv, &server{client: c}) reflection.Register(srv) + m.InitializeServer(srv) - // go func() { - // log.Printf("grpc ran on tcp protocol %s", ADDR) - // log.Fatal(srv.Serve(tcpLn)) - // }() + return &clusterServer{name: name, grpc: srv, listener: ln, client: c}, nil +} - log.Printf("Cake is ready for unix socket protocol: %s", ac.Socket) - log.Println(srv.Serve(ln)) +// shutdownServers gracefully stops the gRPC servers and closes their Aerospike +// clients, draining in-flight RPCs. +func shutdownServers(servers []*clusterServer) { + for _, cs := range servers { + cs.grpc.GracefulStop() + cs.client.Close() + } +} + +func cleanUp(conf map[string]*client.AerospikeConfig, logger *slog.Logger) { + for _, ac := range conf { + if err := os.Remove(ac.Socket); err != nil && !errors.Is(err, os.ErrNotExist) && !errors.Is(err, fs.ErrNotExist) { + logger.Warn("socket was not cleaned up", "socket", ac.Socket, "err", err) + } + } +} + +// aerospikeStats adapts *aero.Client onto metrics.StatsProvider. The Aerospike +// client returns its own error type, so a thin wrapper is needed to satisfy the +// standard-error interface used by the metrics package. +type aerospikeStats struct { + c *aero.Client +} + +func (a aerospikeStats) Stats() (map[string]interface{}, error) { + return a.c.Stats() } func init() { if info, ok := debug.ReadBuildInfo(); ok { for _, kv := range info.Settings { - if kv.Value == "" { - continue - } - switch kv.Key { - case "vcs.revision": + if kv.Key == "vcs.revision" && kv.Value != "" { revision = kv.Value - case "vcs.time": - lastCommit, _ = time.Parse(time.RFC3339, kv.Value) } } } diff --git a/aerospike-connection-manager/pkg/build/usr/lib/systemd/system/aerospike-connection-manager.service b/aerospike-connection-manager/pkg/build/usr/lib/systemd/system/aerospike-connection-manager.service index 6a45f441..4d7ceb7d 100644 --- a/aerospike-connection-manager/pkg/build/usr/lib/systemd/system/aerospike-connection-manager.service +++ b/aerospike-connection-manager/pkg/build/usr/lib/systemd/system/aerospike-connection-manager.service @@ -1,11 +1,11 @@ [Unit] Description=Aerospike Local Daemon Service -Documentation=https://https://github.com/aerospike/php-client/asld +Documentation=https://github.com/aerospike/php-client Wants=network.target After=network-online.target [Service] -ExecStart=/usr/bin/aerospike-connection-manager --config /etc/aerospike-connection-manager/asld.toml +ExecStart=/usr/bin/aerospike-connection-manager -config-file /etc/aerospike-connection-manager/asld.toml [Install] WantedBy=multi-user.target