diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c7db4c1e..1ff51643c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # Change Log ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A) +- (Feature) Inventory Collector - (Feature) (RBAC) Add ArangoPermissionRoleUserBinding handler - (Feature) (Gateway) Use HTTP/2 for the default upstream to ArangoDB (hidden feature `gateway-default-http2`, enabled by default) - (Bugfix) (RBAC) Fix scope evaluation, pool propagation, missing indexes, and add APIRefresh endpoint diff --git a/README.md b/README.md index fae03f048..d659669fc 100644 --- a/README.md +++ b/README.md @@ -202,7 +202,7 @@ Flags: --kubernetes.qps float32 Number of queries per second for k8s API. If set to 0 or less, API calls won't be throttled (default 32) --leader.label.skip Skips Leader Label for the Pod --log.format string Set log format. Allowed values: 'pretty', 'JSON'. If empty, default format is used (default "pretty") - --log.level stringArray Set log levels in format or =. Possible loggers: action, agency, arangod-request, assertion, authz-pool-client, backup-operator, backup-policy-operator, chaos-monkey, cli-utils, crd, deployment, deployment-ci, deployment-reconcile, deployment-resilience, deployment-resources, deployment-storage, deployment-storage-pc, deployment-storage-service, generic-parent-operator, grpc-service, helm, http, inspector, integration-authn-v1, integration-authorization-v1, integration-config-v1, integration-envoy-auth-v3, integration-envoy-auth-v3-impl-auth-bearer, integration-envoy-auth-v3-impl-auth-cookie, integration-envoy-auth-v3-impl-custom-openid, integration-envoy-auth-v3-impl-pass-mode, integration-events-v1, integration-meta-v1, integration-pong-v1, integration-scheduler-v2, integration-shutdown-v1, integration-storage-v1-s3, integration-storage-v2, integrations, k8s-client, kubernetes, kubernetes-access, kubernetes-admission, kubernetes-client, kubernetes-informer, monitor, networking-route-operator, operator, operator-v2, operator-v2-event, operator-v2-worker, panics, permission-role-user-binding-operator, permission-token-operator, permission-token-webhook, platform-chart-operator, platform-pod-shutdown, platform-service-operator, platform-storage-operator, pod_compare, root, root-event-recorder, scheduler-batchjob-operator, scheduler-cronjob-operator, scheduler-deployment-operator, scheduler-pod-operator, scheduler-profile-operator, sidecar, sidecar-authz, sidecar-authz-pool, sidecar-service-client-authentication, webhook (default [info]) + --log.level stringArray Set log levels in format or =. Possible loggers: action, agency, arangod-request, assertion, authz-pool-client, backup-operator, backup-policy-operator, chaos-monkey, cli-utils, collector, crd, deployment, deployment-ci, deployment-reconcile, deployment-resilience, deployment-resources, deployment-storage, deployment-storage-pc, deployment-storage-service, generic-parent-operator, grpc-service, helm, http, inspector, integration-authn-v1, integration-authorization-v1, integration-config-v1, integration-envoy-auth-v3, integration-envoy-auth-v3-impl-auth-bearer, integration-envoy-auth-v3-impl-auth-cookie, integration-envoy-auth-v3-impl-custom-openid, integration-envoy-auth-v3-impl-pass-mode, integration-events-v1, integration-meta-v1, integration-pong-v1, integration-scheduler-v2, integration-shutdown-v1, integration-storage-v1-s3, integration-storage-v2, integrations, k8s-client, kubernetes, kubernetes-access, kubernetes-admission, kubernetes-client, kubernetes-informer, monitor, networking-route-operator, operator, operator-v2, operator-v2-event, operator-v2-worker, panics, permission-role-user-binding-operator, permission-token-operator, permission-token-webhook, platform-chart-operator, platform-pod-shutdown, platform-service-operator, platform-storage-operator, pod_compare, root, root-event-recorder, scheduler-batchjob-operator, scheduler-cronjob-operator, scheduler-deployment-operator, scheduler-pod-operator, scheduler-profile-operator, sidecar, sidecar-authz, sidecar-authz-pool, sidecar-service-client-authentication, webhook (default [info]) --log.sampling If true, operator will try to minimize duplication of logging events (default true) --log.stdout If true, operator will log to the stdout (default true) --memory-limit uint Define memory limit for hard shutdown and the dump of goroutines. Used for testing diff --git a/cmd/lifecycle_post_start.go b/cmd/lifecycle_post_start.go new file mode 100644 index 000000000..5ee15d080 --- /dev/null +++ b/cmd/lifecycle_post_start.go @@ -0,0 +1,62 @@ +// +// DISCLAIMER +// +// Copyright 2016-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package cmd + +import ( + "github.com/spf13/cobra" + + "github.com/arangodb/kube-arangodb/pkg/collect" + "github.com/arangodb/kube-arangodb/pkg/util/cli" +) + +var ( + // cmdLifecyclePostStart is the parent for postStart lifecycle hooks (`lifecycle postStart ...`). + cmdLifecyclePostStart = &cobra.Command{ + Use: "postStart", + RunE: cli.Usage, + Hidden: true, + } + + // cmdLifecyclePostStartCollector runs the collector as a postStart hook (`lifecycle postStart collector`). + cmdLifecyclePostStartCollector = &cobra.Command{ + Use: "collector", + Hidden: true, + } + + // lifecyclePostStartCollectorOptions holds the flags bound to the collector command. + lifecyclePostStartCollectorOptions collect.Options +) + +func init() { + cmdLifecyclePostStartCollector.RunE = cmdLifecyclePostStartCollectorRunE + + f := cmdLifecyclePostStartCollector.Flags() + f.DurationVar(&lifecyclePostStartCollectorOptions.Interval, "interval", collect.DefaultInterval, "Collector retry interval") + f.DurationVar(&lifecyclePostStartCollectorOptions.Timeout, "timeout", collect.DefaultTimeout, "Collector run timeout") + + cmdLifecyclePostStart.AddCommand(cmdLifecyclePostStartCollector) + cmdLifecycle.AddCommand(cmdLifecyclePostStart) +} + +// cmdLifecyclePostStartCollectorRunE delegates to the collector implementation in pkg/collect. +func cmdLifecyclePostStartCollectorRunE(cmd *cobra.Command, _ []string) error { + return collect.PostStart(cmd.Context(), lifecyclePostStartCollectorOptions) +} diff --git a/docs/cli/arangodb_operator.md b/docs/cli/arangodb_operator.md index adbe7c50b..bea2da9fd 100644 --- a/docs/cli/arangodb_operator.md +++ b/docs/cli/arangodb_operator.md @@ -87,7 +87,7 @@ Flags: --kubernetes.qps float32 Number of queries per second for k8s API. If set to 0 or less, API calls won't be throttled (default 32) --leader.label.skip Skips Leader Label for the Pod --log.format string Set log format. Allowed values: 'pretty', 'JSON'. If empty, default format is used (default "pretty") - --log.level stringArray Set log levels in format or =. Possible loggers: action, agency, arangod-request, assertion, authz-pool-client, backup-operator, backup-policy-operator, chaos-monkey, cli-utils, crd, deployment, deployment-ci, deployment-reconcile, deployment-resilience, deployment-resources, deployment-storage, deployment-storage-pc, deployment-storage-service, generic-parent-operator, grpc-service, helm, http, inspector, integration-authn-v1, integration-authorization-v1, integration-config-v1, integration-envoy-auth-v3, integration-envoy-auth-v3-impl-auth-bearer, integration-envoy-auth-v3-impl-auth-cookie, integration-envoy-auth-v3-impl-custom-openid, integration-envoy-auth-v3-impl-pass-mode, integration-events-v1, integration-meta-v1, integration-pong-v1, integration-scheduler-v2, integration-shutdown-v1, integration-storage-v1-s3, integration-storage-v2, integrations, k8s-client, kubernetes, kubernetes-access, kubernetes-admission, kubernetes-client, kubernetes-informer, monitor, networking-route-operator, operator, operator-v2, operator-v2-event, operator-v2-worker, panics, permission-role-user-binding-operator, permission-token-operator, permission-token-webhook, platform-chart-operator, platform-pod-shutdown, platform-service-operator, platform-storage-operator, pod_compare, root, root-event-recorder, scheduler-batchjob-operator, scheduler-cronjob-operator, scheduler-deployment-operator, scheduler-pod-operator, scheduler-profile-operator, sidecar, sidecar-authz, sidecar-authz-pool, sidecar-service-client-authentication, webhook (default [info]) + --log.level stringArray Set log levels in format or =. Possible loggers: action, agency, arangod-request, assertion, authz-pool-client, backup-operator, backup-policy-operator, chaos-monkey, cli-utils, collector, crd, deployment, deployment-ci, deployment-reconcile, deployment-resilience, deployment-resources, deployment-storage, deployment-storage-pc, deployment-storage-service, generic-parent-operator, grpc-service, helm, http, inspector, integration-authn-v1, integration-authorization-v1, integration-config-v1, integration-envoy-auth-v3, integration-envoy-auth-v3-impl-auth-bearer, integration-envoy-auth-v3-impl-auth-cookie, integration-envoy-auth-v3-impl-custom-openid, integration-envoy-auth-v3-impl-pass-mode, integration-events-v1, integration-meta-v1, integration-pong-v1, integration-scheduler-v2, integration-shutdown-v1, integration-storage-v1-s3, integration-storage-v2, integrations, k8s-client, kubernetes, kubernetes-access, kubernetes-admission, kubernetes-client, kubernetes-informer, monitor, networking-route-operator, operator, operator-v2, operator-v2-event, operator-v2-worker, panics, permission-role-user-binding-operator, permission-token-operator, permission-token-webhook, platform-chart-operator, platform-pod-shutdown, platform-service-operator, platform-storage-operator, pod_compare, root, root-event-recorder, scheduler-batchjob-operator, scheduler-cronjob-operator, scheduler-deployment-operator, scheduler-pod-operator, scheduler-profile-operator, sidecar, sidecar-authz, sidecar-authz-pool, sidecar-service-client-authentication, webhook (default [info]) --log.sampling If true, operator will try to minimize duplication of logging events (default true) --log.stdout If true, operator will log to the stdout (default true) --memory-limit uint Define memory limit for hard shutdown and the dump of goroutines. Used for testing diff --git a/pkg/collect/collect.go b/pkg/collect/collect.go new file mode 100644 index 000000000..aa94b125e --- /dev/null +++ b/pkg/collect/collect.go @@ -0,0 +1,191 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +// Package collect implements the collector run by the arangod/gateway postStart lifecycle hook. +// +// The collector is a simple application started by the postStart hook. It runs in the foreground, +// collects the events for the current boot on a fixed interval and stops as soon as a cycle +// succeeds. It is not a daemon - there is no background process to keep alive - it simply runs until +// it is done. A timeout bounds the whole run so it cannot retry indefinitely if it never succeeds. +// +// Collected events are printed to stdout. +// +// Each boot is identified by a unique boot id (see shutdown.BootID), which is attached to every +// event the collector emits so that all activity from a single pod boot can be correlated. +package collect + +import ( + "context" + "fmt" + "os" + "time" + + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/types/known/timestamppb" + + pbEventsV1 "github.com/arangodb/kube-arangodb/integrations/events/v1/definition" + "github.com/arangodb/kube-arangodb/pkg/util/errors" + "github.com/arangodb/kube-arangodb/pkg/util/shutdown" + "github.com/arangodb/kube-arangodb/pkg/version" +) + +const ( + // serviceID identifies the collector as the source of the emitted event. + serviceID = "collector" + + // eventTypeStartup is the type of the event emitted on every pod startup. + eventTypeStartup = "startup" + + // dimensionBootID is the event dimension carrying the unique boot identifier. + dimensionBootID = "bootID" +) + +const ( + // DefaultInterval is the default retry interval between collector cycles. + DefaultInterval = 30 * time.Second + + // DefaultTimeout is the default upper bound for the collector, after which it gives up. + DefaultTimeout = 30 * time.Minute +) + +// Options configures a collector run. +type Options struct { + // Interval is the retry interval between collector cycles. Defaults to DefaultInterval. + Interval time.Duration + + // Timeout bounds the whole run. When the timeout elapses before a cycle succeeds the collector + // gives up and returns. A non-positive value falls back to DefaultTimeout. + Timeout time.Duration +} + +// interval returns the configured retry interval, falling back to DefaultInterval. +func (o Options) interval() time.Duration { + if o.Interval <= 0 { + return DefaultInterval + } + return o.Interval +} + +// timeout returns the configured run timeout, falling back to DefaultTimeout. +func (o Options) timeout() time.Duration { + if o.Timeout <= 0 { + return DefaultTimeout + } + return o.Timeout +} + +// PostStart is the entrypoint for the postStart collector lifecycle hook. +// +// It runs the collector loop in the foreground and returns once a cycle succeeds, the context is +// cancelled, or the timeout elapses. It never spawns a background process. +func PostStart(ctx context.Context, opts Options) error { + ctx, cancel := context.WithTimeout(ctx, opts.timeout()) + defer cancel() + + return run(ctx, opts) +} + +// run executes the collector loop. A single failing cycle never aborts the run, it just waits for +// the next interval and retries, until a cycle succeeds or the context is closed (cancelled or timed +// out). +func run(ctx context.Context, opts Options) error { + // bootID is stable for the lifetime of this process and identifies the current boot. + bootID := shutdown.BootID() + + // created is the process boot time, injected into every collected event so all events share a + // single, stable timestamp regardless of which cycle produced them. It is trimmed to seconds to + // keep cross-platform compatibility with the events integration. + created := shutdown.BootTime().Truncate(time.Second) + + logger.Str("bootID", bootID).Info("Starting arangodb-operator collector (%s), version %s build %s", + version.GetVersionV1().Edition.Title(), version.GetVersionV1().Version, version.GetVersionV1().Build) + + t := time.NewTicker(opts.interval()) + defer t.Stop() + + for { + if err := collect(bootID, created); err != nil { + logger.Err(err).Str("bootID", bootID).Warn("Collector cycle failed, will retry") + } else { + logger.Str("bootID", bootID).Info("Collector finished") + return nil + } + + select { + case <-ctx.Done(): + logger.Err(ctx.Err()).Str("bootID", bootID).Warn("Collector stopped before completion") + return ctx.Err() + case <-t.C: + } + } +} + +// collect performs a single collection cycle for the given boot: it runs every registered collector, +// each pushing its metrics into a shared collector, waits until all of them have completed, builds a +// single startup event whose body is the collected metrics and prints it to stdout. The event is +// tagged with the boot id and the start timestamp so it can be correlated to a single pod boot. +func collect(bootID string, created time.Time) error { + metrics, err := GetCollector().Collect() + if err != nil { + return err + } + + event := buildEvent(metrics, bootID, created) + + if err := print(event); err != nil { + return err + } + + logger.Str("bootID", bootID).Int("metrics", len(metrics)).Debug("Collected metrics") + return nil +} + +// buildEvent assembles the startup event from the collected metrics, tagging it with the boot id and +// the start timestamp. +func buildEvent(metrics []Metric, bootID string, created time.Time) *pbEventsV1.Event { + body := make(map[string]float32, len(metrics)) + for _, m := range metrics { + body[m.K] = m.V + } + + return &pbEventsV1.Event{ + Type: eventTypeStartup, + ServiceId: serviceID, + Created: timestamppb.New(created), + Dimensions: map[string]string{ + dimensionBootID: bootID, + }, + Body: body, + } +} + +// print writes the event to stdout as a single JSON line. +func print(event *pbEventsV1.Event) error { + data, err := protojson.Marshal(event) + if err != nil { + return errors.Wrapf(err, "unable to marshal event") + } + + if _, err := fmt.Fprintln(os.Stdout, string(data)); err != nil { + return errors.Wrapf(err, "unable to write event") + } + + return nil +} diff --git a/pkg/collect/collect_test.go b/pkg/collect/collect_test.go new file mode 100644 index 000000000..c255be20f --- /dev/null +++ b/pkg/collect/collect_test.go @@ -0,0 +1,107 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package collect + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/arangodb/kube-arangodb/pkg/util" +) + +// fakeCollector is a test ECollector pushing a fixed set of metrics, or returning an error. +type fakeCollector struct { + metrics []Metric + err error +} + +func (f fakeCollector) CollectEvents(out util.Pusher[Metric]) error { + if f.err != nil { + return f.err + } + out.Push(f.metrics...) + return nil +} + +func TestRegistry_Collect(t *testing.T) { + c := NewCollector[Metric]() + + c.Register(fakeCollector{metrics: []Metric{{K: "a", V: 1}}}) + c.Register(fakeCollector{metrics: []Metric{{K: "b", V: 2}, {K: "c", V: 3}}}) + + metrics, err := c.Collect() + require.NoError(t, err) + + values := map[string]float32{} + for _, m := range metrics { + values[m.K] = m.V + } + require.Equal(t, map[string]float32{"a": 1, "b": 2, "c": 3}, values) +} + +func TestRegistry_CollectEmpty(t *testing.T) { + c := NewCollector[Metric]() + + metrics, err := c.Collect() + require.NoError(t, err) + require.Empty(t, metrics) +} + +func TestRegistry_CollectError(t *testing.T) { + c := NewCollector[Metric]() + + c.Register(fakeCollector{metrics: []Metric{{K: "a", V: 1}}}) + c.Register(fakeCollector{err: errBoom}) + + metrics, err := c.Collect() + require.ErrorIs(t, err, errBoom) + require.Nil(t, metrics) +} + +func TestBuildEvent(t *testing.T) { + created := time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) + + event := buildEvent([]Metric{{K: "cpu", V: 4}, {K: "memory", V: 1024}}, "boot-123", created) + + require.Equal(t, eventTypeStartup, event.GetType()) + require.Equal(t, serviceID, event.GetServiceId()) + require.Equal(t, created, event.GetCreated().AsTime()) + require.Equal(t, "boot-123", event.GetDimensions()[dimensionBootID]) + require.Equal(t, map[string]float32{"cpu": 4, "memory": 1024}, event.GetBody()) +} + +func TestBuildEvent_NoMetrics(t *testing.T) { + created := time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) + + event := buildEvent(nil, "boot-123", created) + + require.Equal(t, eventTypeStartup, event.GetType()) + require.Equal(t, "boot-123", event.GetDimensions()[dimensionBootID]) + require.Empty(t, event.GetBody()) +} + +var errBoom = boomError("boom") + +type boomError string + +func (e boomError) Error() string { return string(e) } diff --git a/pkg/collect/collector.go b/pkg/collect/collector.go new file mode 100644 index 000000000..dfcd3956c --- /dev/null +++ b/pkg/collect/collector.go @@ -0,0 +1,100 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package collect + +import ( + "sync" + + "github.com/arangodb/kube-arangodb/pkg/util" +) + +// Metric is a single event body value: a key with a float value. Collectors push metrics, which the +// collector assembles into the body of the emitted event. +type Metric = util.KV[string, float32] + +// ECollector is a single source of values of type T. It pushes every value it produces to the +// provided pusher and returns an error if it failed. Returning an error fails the whole collection, +// which is then retried on the next interval - so a collector should only return an error when it +// could not collect, not when it simply has nothing to push. +type ECollector[T any] interface { + CollectEvents(out util.Pusher[T]) error +} + +// Collector is a registry of ECollector[T]. Collectors register once (typically from an init +// function) and Collect fans them into a single list. +type Collector[T any] interface { + // Register adds a collector to the registry. + Register(c ECollector[T]) + + // Collect runs every registered collector and returns the aggregated values, or the first + // collector error. + Collect() ([]T, error) +} + +// NewCollector creates an empty Collector registry. +func NewCollector[T any]() Collector[T] { + return &collector[T]{} +} + +var registry = NewCollector[Metric]() + +// GetCollector returns the global registry of event metric collectors. +func GetCollector() Collector[Metric] { + return registry +} + +type collector[T any] struct { + lock sync.Mutex + + collectors []ECollector[T] +} + +func (c *collector[T]) Register(e ECollector[T]) { + c.lock.Lock() + defer c.lock.Unlock() + + c.collectors = append(c.collectors, e) +} + +// Collect runs every registered collector concurrently in the background, each pushing its values +// into a shared collector. It waits until all of them have completed, then returns the aggregated +// list (or the first collector error). +func (c *collector[T]) Collect() ([]T, error) { + c.lock.Lock() + defer c.lock.Unlock() + + out := util.NewCollector[T]() + + for _, e := range c.collectors { + e := e + if err := out.Run(func(p util.Pusher[T]) error { + return e.CollectEvents(p) + }); err != nil { + return nil, err + } + } + + if err := out.Done(); err != nil { + return nil, err + } + + return out.Collect(), nil +} diff --git a/pkg/collect/collector_cgroup_linux.go b/pkg/collect/collector_cgroup_linux.go new file mode 100644 index 000000000..984b507c5 --- /dev/null +++ b/pkg/collect/collector_cgroup_linux.go @@ -0,0 +1,214 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build linux + +package collect + +import ( + "os" + "path/filepath" + "strconv" + goStrings "strings" + + "github.com/arangodb/kube-arangodb/pkg/util" + "github.com/arangodb/kube-arangodb/pkg/util/errors" +) + +// The resource collector in collector_resource.go reports host-level values (runtime.NumCPU and +// /proc/meminfo). Inside a container those over-report: they reflect the node, not the cgroup limit +// the customer was actually allocated. This collector reports the cgroup-enforced limits instead, +// supporting both cgroup v1 (per-controller hierarchy) and cgroup v2 (unified hierarchy). +// +// The cgroup filesystem is a Linux-only kernel interface, so this collector is compiled and +// registered on Linux only. See collector_cgroup_other.go for the no-op on other platforms. +// +// When no limit is enforced (unlimited) or the cgroup files are not present (e.g. running outside a +// container), the corresponding metric is simply not pushed - this is not an error, the collector +// just has nothing to report. + +const ( + // metricCPULimit is the number of CPUs the cgroup allows (fractional, e.g. quota/period). + metricCPULimit = "system_cpu_limit" + + // metricMemoryLimit is the cgroup memory limit in GB. + metricMemoryLimit = "system_memory_limit" + + // defaultCgroupRoot is the standard mount point of the cgroup filesystem. + defaultCgroupRoot = "/sys/fs/cgroup" + + // unlimitedMemoryThreshold guards against the cgroup v1 "no limit" sentinel, which is reported + // as a page-aligned near-max int64 (e.g. 9223372036854771712). Any value at or above this is + // treated as unlimited rather than a real limit. + unlimitedMemoryThreshold = uint64(1) << 62 +) + +func init() { + GetCollector().Register(cgroupCollector{root: defaultCgroupRoot}) +} + +// cgroupCollector pushes the cgroup-enforced CPU and memory limits as event body metrics. +type cgroupCollector struct { + // root is the cgroup filesystem mount point. Overridable for testing. + root string +} + +// CollectEvents pushes the cgroup CPU limit (in CPUs) and memory limit (in GB), when enforced. +func (c cgroupCollector) CollectEvents(out util.Pusher[Metric]) error { + root := c.root + if root == "" { + root = defaultCgroupRoot + } + + v2 := isCgroupV2(root) + + cpu, ok, err := cgroupCPULimit(root, v2) + if err != nil { + return err + } + if ok { + out.Push(Metric{K: metricCPULimit, V: cpu}) + } + + mem, ok, err := cgroupMemoryLimit(root, v2) + if err != nil { + return err + } + if ok { + out.Push(Metric{K: metricMemoryLimit, V: float32(mem) / bytesPerGB}) + } + + return nil +} + +// isCgroupV2 reports whether the cgroup mount is the unified (v2) hierarchy. The unified hierarchy +// is identified by the presence of the cgroup.controllers file at the root. +func isCgroupV2(root string) bool { + _, err := os.Stat(filepath.Join(root, "cgroup.controllers")) + return err == nil +} + +// cgroupCPULimit returns the number of CPUs the cgroup permits (quota/period). ok is false when no +// limit is enforced or the relevant files are absent (running outside a container). +func cgroupCPULimit(root string, v2 bool) (float32, bool, error) { + if v2 { + // cgroup v2: cpu.max holds " ", or "max " when unlimited. + line, ok, err := readCgroupFile(filepath.Join(root, "cpu.max")) + if err != nil || !ok { + return 0, false, err + } + + fields := goStrings.Fields(line) + if len(fields) < 2 { + return 0, false, errors.Errorf("unexpected cpu.max content: %q", line) + } + if fields[0] == "max" { + return 0, false, nil + } + + quota, err := strconv.ParseInt(fields[0], 10, 64) + if err != nil { + return 0, false, errors.Wrapf(err, "unable to parse cpu.max quota: %q", line) + } + period, err := strconv.ParseInt(fields[1], 10, 64) + if err != nil { + return 0, false, errors.Wrapf(err, "unable to parse cpu.max period: %q", line) + } + if period <= 0 { + return 0, false, nil + } + return float32(quota) / float32(period), true, nil + } + + // cgroup v1: cpu.cfs_quota_us / cpu.cfs_period_us. A quota of -1 means unlimited. + quotaStr, ok, err := readCgroupFile(filepath.Join(root, "cpu", "cpu.cfs_quota_us")) + if err != nil || !ok { + return 0, false, err + } + quota, err := strconv.ParseInt(goStrings.TrimSpace(quotaStr), 10, 64) + if err != nil { + return 0, false, errors.Wrapf(err, "unable to parse cpu.cfs_quota_us: %q", quotaStr) + } + if quota <= 0 { + return 0, false, nil + } + + periodStr, ok, err := readCgroupFile(filepath.Join(root, "cpu", "cpu.cfs_period_us")) + if err != nil || !ok { + return 0, false, err + } + period, err := strconv.ParseInt(goStrings.TrimSpace(periodStr), 10, 64) + if err != nil { + return 0, false, errors.Wrapf(err, "unable to parse cpu.cfs_period_us: %q", periodStr) + } + if period <= 0 { + return 0, false, nil + } + + return float32(quota) / float32(period), true, nil +} + +// cgroupMemoryLimit returns the cgroup memory limit in bytes. ok is false when no limit is enforced +// or the relevant files are absent. +func cgroupMemoryLimit(root string, v2 bool) (uint64, bool, error) { + var file string + if v2 { + file = filepath.Join(root, "memory.max") + } else { + file = filepath.Join(root, "memory", "memory.limit_in_bytes") + } + + line, ok, err := readCgroupFile(file) + if err != nil || !ok { + return 0, false, err + } + + line = goStrings.TrimSpace(line) + + // cgroup v2 reports "max" for no limit. + if line == "max" { + return 0, false, nil + } + + limit, err := strconv.ParseUint(line, 10, 64) + if err != nil { + return 0, false, errors.Wrapf(err, "unable to parse memory limit: %q", line) + } + + // cgroup v1 reports a near-max sentinel instead of "max". + if limit >= unlimitedMemoryThreshold { + return 0, false, nil + } + + return limit, true, nil +} + +// readCgroupFile reads a cgroup file, returning its trimmed content. ok is false (without error) +// when the file does not exist, so callers can treat a missing cgroup file as "no limit reported". +func readCgroupFile(path string) (string, bool, error) { + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return "", false, nil + } + return "", false, errors.Wrapf(err, "unable to read %q", path) + } + return goStrings.TrimSpace(string(data)), true, nil +} diff --git a/pkg/collect/collector_cgroup_other.go b/pkg/collect/collector_cgroup_other.go new file mode 100644 index 000000000..10841b32d --- /dev/null +++ b/pkg/collect/collector_cgroup_other.go @@ -0,0 +1,28 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build !linux + +package collect + +// The cgroup collector (collector_cgroup_linux.go) reads the Linux cgroup filesystem +// (/sys/fs/cgroup), a Linux-only kernel interface. On non-Linux platforms it is intentionally not +// compiled or registered, so the package still builds for local development and tooling. No cgroup +// resource metrics are emitted off Linux. diff --git a/pkg/collect/collector_cgroup_test.go b/pkg/collect/collector_cgroup_test.go new file mode 100644 index 000000000..4aec6a479 --- /dev/null +++ b/pkg/collect/collector_cgroup_test.go @@ -0,0 +1,117 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build linux + +package collect + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/arangodb/kube-arangodb/pkg/util" +) + +// writeFile writes content to path, creating parent directories as needed. +func writeFile(t *testing.T, path, content string) { + t.Helper() + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o755)) + require.NoError(t, os.WriteFile(path, []byte(content), 0o644)) +} + +// collectValues runs a collector and returns its pushed metrics as a map. +func collectValues(t *testing.T, c ECollector[Metric]) map[string]float32 { + t.Helper() + out := util.NewCollector[Metric]() + require.NoError(t, c.CollectEvents(out)) + require.NoError(t, out.Done()) + + values := map[string]float32{} + for _, m := range out.Collect() { + values[m.K] = m.V + } + return values +} + +func TestCgroupCollector_V2(t *testing.T) { + root := t.TempDir() + // Presence of cgroup.controllers marks the unified (v2) hierarchy. + writeFile(t, filepath.Join(root, "cgroup.controllers"), "cpu memory") + writeFile(t, filepath.Join(root, "cpu.max"), "50000 100000\n") // 0.5 CPUs + writeFile(t, filepath.Join(root, "memory.max"), "2147483648\n") // 2 GiB + + values := collectValues(t, cgroupCollector{root: root}) + + require.InDelta(t, 0.5, values[metricCPULimit], 1e-6) + require.InDelta(t, 2.0, values[metricMemoryLimit], 1e-6) +} + +func TestCgroupCollector_V2_Unlimited(t *testing.T) { + root := t.TempDir() + writeFile(t, filepath.Join(root, "cgroup.controllers"), "cpu memory") + writeFile(t, filepath.Join(root, "cpu.max"), "max 100000\n") + writeFile(t, filepath.Join(root, "memory.max"), "max\n") + + values := collectValues(t, cgroupCollector{root: root}) + + // Unlimited means nothing is pushed. + _, hasCPU := values[metricCPULimit] + _, hasMem := values[metricMemoryLimit] + require.False(t, hasCPU) + require.False(t, hasMem) +} + +func TestCgroupCollector_V1(t *testing.T) { + root := t.TempDir() + // No cgroup.controllers file -> v1 hierarchy. + writeFile(t, filepath.Join(root, "cpu", "cpu.cfs_quota_us"), "150000\n") // 1.5 CPUs + writeFile(t, filepath.Join(root, "cpu", "cpu.cfs_period_us"), "100000\n") + writeFile(t, filepath.Join(root, "memory", "memory.limit_in_bytes"), "1073741824\n") // 1 GiB + + values := collectValues(t, cgroupCollector{root: root}) + + require.InDelta(t, 1.5, values[metricCPULimit], 1e-6) + require.InDelta(t, 1.0, values[metricMemoryLimit], 1e-6) +} + +func TestCgroupCollector_V1_Unlimited(t *testing.T) { + root := t.TempDir() + writeFile(t, filepath.Join(root, "cpu", "cpu.cfs_quota_us"), "-1\n") + writeFile(t, filepath.Join(root, "cpu", "cpu.cfs_period_us"), "100000\n") + // The v1 "no limit" sentinel. + writeFile(t, filepath.Join(root, "memory", "memory.limit_in_bytes"), "9223372036854771712\n") + + values := collectValues(t, cgroupCollector{root: root}) + + _, hasCPU := values[metricCPULimit] + _, hasMem := values[metricMemoryLimit] + require.False(t, hasCPU) + require.False(t, hasMem) +} + +func TestCgroupCollector_Absent(t *testing.T) { + // An empty root has none of the cgroup files: the collector must succeed and push nothing, + // rather than fail the whole collection (e.g. when running outside a container). + values := collectValues(t, cgroupCollector{root: t.TempDir()}) + require.Empty(t, values) +} diff --git a/pkg/collect/collector_filesystem_linux.go b/pkg/collect/collector_filesystem_linux.go new file mode 100644 index 000000000..0c9f52f05 --- /dev/null +++ b/pkg/collect/collector_filesystem_linux.go @@ -0,0 +1,83 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build linux + +package collect + +import ( + "golang.org/x/sys/unix" + + "github.com/arangodb/kube-arangodb/pkg/util" + "github.com/arangodb/kube-arangodb/pkg/util/errors" +) + +// filesystemCollector reports the total and available capacity of the filesystem backing a given +// path, obtained from statfs. This is the disk-usage half of the resource inventory: how much +// storage the container has, and how much of it is still free. +// +// statfs is compiled on Linux only here; see collector_filesystem_other.go for the no-op on other +// platforms. + +const ( + // metricFSTotal is the total capacity of the filesystem under the path, in GB. + metricFSTotal = "system_fs_total" + + // metricFSFree is the space available to an unprivileged user on that filesystem, in GB. + metricFSFree = "system_fs_free" + + // defaultFilesystemPath is the path whose backing filesystem is measured by default. The root + // filesystem is a safe default; a real deployment would point this at the data volume. + defaultFilesystemPath = "/" +) + +func init() { + GetCollector().Register(filesystemCollector{path: defaultFilesystemPath}) +} + +// filesystemCollector pushes the total and free capacity of the filesystem backing path. +type filesystemCollector struct { + // path selects the filesystem to measure (any path on the target mount). Overridable for testing. + path string +} + +// CollectEvents pushes the total and available capacity (in GB) of the filesystem backing the path. +func (f filesystemCollector) CollectEvents(out util.Pusher[Metric]) error { + path := f.path + if path == "" { + path = defaultFilesystemPath + } + + var st unix.Statfs_t + if err := unix.Statfs(path, &st); err != nil { + return errors.Wrapf(err, "unable to statfs %q", path) + } + + // Bsize is the fundamental block size; Blocks is the total number of blocks and Bavail the + // blocks available to an unprivileged user. Multiplying by the block size yields bytes. + blockSize := uint64(st.Bsize) + total := st.Blocks * blockSize + free := st.Bavail * blockSize + + out.Push(Metric{K: metricFSTotal, V: float32(total) / bytesPerGB}) + out.Push(Metric{K: metricFSFree, V: float32(free) / bytesPerGB}) + + return nil +} diff --git a/pkg/collect/collector_filesystem_other.go b/pkg/collect/collector_filesystem_other.go new file mode 100644 index 000000000..748d796e0 --- /dev/null +++ b/pkg/collect/collector_filesystem_other.go @@ -0,0 +1,28 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build !linux + +package collect + +// The filesystem collector (collector_filesystem_linux.go) uses the statfs syscall. To keep the +// build tag scheme uniform with the cgroup collector and to avoid pulling platform-specific syscall +// types into non-Linux builds, it is compiled and registered on Linux only. On non-Linux platforms +// no filesystem metrics are emitted, so the package still builds for local development and tooling. diff --git a/pkg/collect/collector_filesystem_test.go b/pkg/collect/collector_filesystem_test.go new file mode 100644 index 000000000..d815dd45f --- /dev/null +++ b/pkg/collect/collector_filesystem_test.go @@ -0,0 +1,48 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build linux + +package collect + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/arangodb/kube-arangodb/pkg/util" +) + +func TestFilesystemCollector(t *testing.T) { + // The temp dir is backed by a real filesystem, so statfs must return a positive total and a + // free value that never exceeds the total. + values := collectValues(t, filesystemCollector{path: t.TempDir()}) + + require.Greater(t, values[metricFSTotal], float32(0)) + require.GreaterOrEqual(t, values[metricFSFree], float32(0)) + require.LessOrEqual(t, values[metricFSFree], values[metricFSTotal]) +} + +func TestFilesystemCollector_MissingPath(t *testing.T) { + // statfs on a non-existent path fails, and that failure propagates so the collection is retried. + out := util.NewCollector[Metric]() + err := filesystemCollector{path: "/this/path/does/not/exist/anywhere"}.CollectEvents(out) + require.Error(t, err) +} diff --git a/pkg/collect/collector_resource_linux.go b/pkg/collect/collector_resource_linux.go new file mode 100644 index 000000000..75c8c12ed --- /dev/null +++ b/pkg/collect/collector_resource_linux.go @@ -0,0 +1,98 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build linux + +package collect + +import ( + "os" + "runtime" + "strconv" + goStrings "strings" + + "github.com/arangodb/kube-arangodb/pkg/util" + "github.com/arangodb/kube-arangodb/pkg/util/errors" +) + +// The resource collector reports host-level CPU and memory. Total memory is read from /proc/meminfo, +// a Linux-only interface, so this collector is compiled and registered on Linux only. See +// collector_resource_other.go for the no-op on other platforms. + +const ( + // metricCPU is the number of CPUs available to the process. + metricCPU = "system_cpu" + + // metricMemory is the total system memory in GB. + metricMemory = "system_memory" + + // bytesPerGB is the number of bytes in one gigabyte. + bytesPerGB = 1024 * 1024 * 1024 +) + +func init() { + GetCollector().Register(resourceCollector{}) +} + +// resourceCollector pushes the available CPU and memory as event body metrics. +type resourceCollector struct{} + +// CollectEvents pushes the CPU count and total memory (in GB). +func (resourceCollector) CollectEvents(out util.Pusher[Metric]) error { + out.Push(Metric{K: metricCPU, V: float32(runtime.NumCPU())}) + + mem, err := totalMemory() + if err != nil { + return err + } + + out.Push(Metric{K: metricMemory, V: float32(mem) / bytesPerGB}) + + return nil +} + +// totalMemory returns the total system memory in bytes, read from /proc/meminfo. +func totalMemory() (uint64, error) { + data, err := os.ReadFile("/proc/meminfo") + if err != nil { + return 0, errors.Wrapf(err, "unable to read /proc/meminfo") + } + + for _, line := range goStrings.Split(string(data), "\n") { + if !goStrings.HasPrefix(line, "MemTotal:") { + continue + } + + // MemTotal is reported in kB, e.g. "MemTotal: 16384000 kB". + fields := goStrings.Fields(line) + if len(fields) < 2 { + return 0, errors.Errorf("unexpected MemTotal line: %q", line) + } + + kb, err := strconv.ParseUint(fields[1], 10, 64) + if err != nil { + return 0, errors.Wrapf(err, "unable to parse MemTotal: %q", line) + } + + return kb * 1024, nil + } + + return 0, errors.Errorf("MemTotal not found in /proc/meminfo") +} diff --git a/pkg/collect/collector_resource_linux_test.go b/pkg/collect/collector_resource_linux_test.go new file mode 100644 index 000000000..ce36750b7 --- /dev/null +++ b/pkg/collect/collector_resource_linux_test.go @@ -0,0 +1,53 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build linux + +package collect + +import ( + "runtime" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/arangodb/kube-arangodb/pkg/util" +) + +func TestResourceCollector(t *testing.T) { + out := util.NewCollector[Metric]() + + require.NoError(t, resourceCollector{}.CollectEvents(out)) + require.NoError(t, out.Done()) + + values := map[string]float32{} + for _, m := range out.Collect() { + values[m.K] = m.V + } + + require.Equal(t, float32(runtime.NumCPU()), values[metricCPU]) + require.Greater(t, values[metricMemory], float32(0)) +} + +func TestTotalMemory(t *testing.T) { + mem, err := totalMemory() + require.NoError(t, err) + require.Greater(t, mem, uint64(0)) +} diff --git a/pkg/collect/collector_resource_other.go b/pkg/collect/collector_resource_other.go new file mode 100644 index 000000000..518d8df2c --- /dev/null +++ b/pkg/collect/collector_resource_other.go @@ -0,0 +1,28 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +//go:build !linux + +package collect + +// The resource collector (collector_resource_linux.go) reads total memory from /proc/meminfo, a +// Linux-only interface. On non-Linux platforms it is intentionally not compiled or registered, so +// the package still builds for local development and tooling. No host resource metrics are emitted +// off Linux. diff --git a/pkg/collect/logger.go b/pkg/collect/logger.go new file mode 100644 index 000000000..24431f21b --- /dev/null +++ b/pkg/collect/logger.go @@ -0,0 +1,26 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package collect + +import "github.com/arangodb/kube-arangodb/pkg/logging" + +// logger is the package-scoped logger used by the collector. +var logger = logging.Global().RegisterAndGetLogger("collector", logging.Info) diff --git a/pkg/deployment/features/collector.go b/pkg/deployment/features/collector.go new file mode 100644 index 000000000..5ff4f503c --- /dev/null +++ b/pkg/deployment/features/collector.go @@ -0,0 +1,37 @@ +// +// DISCLAIMER +// +// Copyright 2016-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package features + +func init() { + registerFeature(collector) +} + +var collector = &feature{ + name: "collector", + description: "Defines if the collector postStart lifecycle hook is enabled on members", + enterpriseRequired: false, + enabledByDefault: false, + hidden: true, +} + +func Collector() Feature { + return collector +} diff --git a/pkg/deployment/resources/pod_creator_arangod.go b/pkg/deployment/resources/pod_creator_arangod.go index 9e576881b..47d58bc66 100644 --- a/pkg/deployment/resources/pod_creator_arangod.go +++ b/pkg/deployment/resources/pod_creator_arangod.go @@ -281,10 +281,23 @@ func (a *ArangoDContainer) GetResourceRequirementsDefaultScale() float64 { } func (a *ArangoDContainer) GetLifecycle() (*core.Lifecycle, error) { + var lifecycle *core.Lifecycle + var err error + if features.GracefulShutdown().Enabled() { - return k8sutil.NewLifecyclePort() + lifecycle, err = k8sutil.NewLifecyclePort() + } else { + lifecycle, err = k8sutil.NewLifecycleFinalizers() + } + if err != nil { + return nil, err + } + + if features.Collector().Enabled() { + lifecycle.PostStart = k8sutil.NewCollectorPostStartHandler() } - return k8sutil.NewLifecycleFinalizers() + + return lifecycle, nil } func (a *ArangoDContainer) GetImagePullPolicy() core.PullPolicy { diff --git a/pkg/deployment/resources/pod_creator_gateway_container.go b/pkg/deployment/resources/pod_creator_gateway_container.go index 87a59133e..23fef6998 100644 --- a/pkg/deployment/resources/pod_creator_gateway_container.go +++ b/pkg/deployment/resources/pod_creator_gateway_container.go @@ -24,6 +24,7 @@ import ( core "k8s.io/api/core/v1" shared "github.com/arangodb/kube-arangodb/pkg/apis/shared" + "github.com/arangodb/kube-arangodb/pkg/deployment/features" utilConstants "github.com/arangodb/kube-arangodb/pkg/util/constants" "github.com/arangodb/kube-arangodb/pkg/util/k8sutil" "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/interfaces" @@ -110,7 +111,16 @@ func (a *MemberGatewayContainer) GetResourceRequirementsDefaultScale() float64 { } func (a *MemberGatewayContainer) GetLifecycle() (*core.Lifecycle, error) { - return k8sutil.NewLifecycleFinalizers() + lifecycle, err := k8sutil.NewLifecycleFinalizers() + if err != nil { + return nil, err + } + + if features.Collector().Enabled() { + lifecycle.PostStart = k8sutil.NewCollectorPostStartHandler() + } + + return lifecycle, nil } func (a *MemberGatewayContainer) GetImagePullPolicy() core.PullPolicy { diff --git a/pkg/util/collector.go b/pkg/util/collector.go new file mode 100644 index 000000000..b389749d9 --- /dev/null +++ b/pkg/util/collector.go @@ -0,0 +1,140 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package util + +import ( + "errors" + "sync" + + "golang.org/x/sync/errgroup" +) + +// ErrCollectorDone is returned by Collector.Run when the collector has already been closed via Done. +var ErrCollectorDone = errors.New("collector is done") + +// Pusher is the producer side of a Collector: it accepts items pushed from one or more producers. +type Pusher[T any] interface { + // Push appends items to the collector. It is safe to call from multiple goroutines. + Push(items ...T) +} + +// Collector aggregates items pushed from one or more producers into a single list. +// +// Producers either Push items directly or are launched in the background with Run. Done waits for +// every background producer to finish and returns the first error, after which Collect returns the +// aggregated list. +// +// Once Done has been called the collector is closed: Run reports ErrCollectorDone and no further +// producers can be started. Done and Collect block until the producers have actually finished, so +// they never return a partial result. +type Collector[T any] interface { + Pusher[T] + + // Run launches fn in the background, passing the collector as its pusher. The error returned by fn + // is reported by Done. Run returns ErrCollectorDone if the collector has already been closed. + Run(fn func(p Pusher[T]) error) error + + // Done closes the collector, waits for every background producer to finish and returns the first + // error. It is idempotent - it only ever closes once and repeated calls return the same error. + Done() error + + // Collect returns all pushed items. It blocks until Done has completed. + Collect() []T +} + +// NewCollector creates a Collector. A background goroutine drains pushed items until Done is called, +// so Push never blocks and is safe to call concurrently. +func NewCollector[T any]() Collector[T] { + c := &collector[T]{ + in: make(chan T), + drained: make(chan struct{}), + } + + go func() { + for v := range c.in { + c.items = append(c.items, v) + } + close(c.drained) + }() + + return c +} + +type collector[T any] struct { + in chan T + drained chan struct{} + + g errgroup.Group + + lock sync.Mutex + closed bool + + once sync.Once + err error + + // items is only ever touched by the draining goroutine, and read by Collect after drained is closed. + items []T +} + +func (c *collector[T]) Push(items ...T) { + for _, i := range items { + c.in <- i + } +} + +func (c *collector[T]) Run(fn func(p Pusher[T]) error) error { + c.lock.Lock() + defer c.lock.Unlock() + + if c.closed { + return ErrCollectorDone + } + + c.g.Go(func() error { + return fn(c) + }) + + return nil +} + +func (c *collector[T]) Done() error { + c.once.Do(func() { + // Close the collector so no further producers can be started. + c.lock.Lock() + c.closed = true + c.lock.Unlock() + + // Wait for every background producer to finish pushing. + c.err = c.g.Wait() + + // No more items will be pushed - close the input and wait for the drain to complete. + close(c.in) + <-c.drained + }) + + return c.err +} + +func (c *collector[T]) Collect() []T { + // Block until the drain has completed, so a partial result is never returned. + <-c.drained + return c.items +} diff --git a/pkg/util/collector_test.go b/pkg/util/collector_test.go new file mode 100644 index 000000000..f113576fe --- /dev/null +++ b/pkg/util/collector_test.go @@ -0,0 +1,200 @@ +// +// DISCLAIMER +// +// Copyright 2024-2026 ArangoDB GmbH, Cologne, Germany +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// Copyright holder is ArangoDB GmbH, Cologne, Germany +// + +package util + +import ( + "sort" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestCollector_Empty(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Done()) + require.Empty(t, c.Collect()) +} + +func TestCollector_DirectPush(t *testing.T) { + c := NewCollector[int]() + + c.Push(1) + c.Push(2, 3) + + require.NoError(t, c.Done()) + + got := c.Collect() + sort.Ints(got) + require.Equal(t, []int{1, 2, 3}, got) +} + +func TestCollector_Run(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Run(func(p Pusher[int]) error { + p.Push(1, 2) + return nil + })) + require.NoError(t, c.Run(func(p Pusher[int]) error { + p.Push(3, 4) + return nil + })) + + require.NoError(t, c.Done()) + + got := c.Collect() + sort.Ints(got) + require.Equal(t, []int{1, 2, 3, 4}, got) +} + +func TestCollector_RunConcurrent(t *testing.T) { + const producers = 16 + const perProducer = 64 + + c := NewCollector[int]() + + for i := 0; i < producers; i++ { + require.NoError(t, c.Run(func(p Pusher[int]) error { + for j := 0; j < perProducer; j++ { + p.Push(1) + } + return nil + })) + } + + require.NoError(t, c.Done()) + + got := c.Collect() + require.Len(t, got, producers*perProducer) + + sum := 0 + for _, v := range got { + sum += v + } + require.Equal(t, producers*perProducer, sum) +} + +func TestCollector_Done_ReturnsFirstError(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Run(func(p Pusher[int]) error { + p.Push(1) + return nil + })) + require.NoError(t, c.Run(func(p Pusher[int]) error { + return errTest + })) + + require.ErrorIs(t, c.Done(), errTest) + + // Even on error, the events that were pushed are still collected. + require.Equal(t, []int{1}, c.Collect()) +} + +func TestCollector_Done_Idempotent(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Run(func(p Pusher[int]) error { + return errTest + })) + + // Repeated Done calls return the same error and do not panic (double close). + require.ErrorIs(t, c.Done(), errTest) + require.ErrorIs(t, c.Done(), errTest) + require.ErrorIs(t, c.Done(), errTest) +} + +func TestCollector_RunAfterDone(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Done()) + + err := c.Run(func(p Pusher[int]) error { + p.Push(1) + return nil + }) + require.ErrorIs(t, err, ErrCollectorDone) + + // The rejected producer did not push anything. + require.Empty(t, c.Collect()) +} + +func TestCollector_CollectBlocksUntilDone(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Run(func(p Pusher[int]) error { + p.Push(1, 2) + return nil + })) + + collected := make(chan []int, 1) + go func() { + collected <- c.Collect() + }() + + // Collect must not return before Done is called. + select { + case <-collected: + require.Fail(t, "Collect returned before Done") + case <-time.After(50 * time.Millisecond): + } + + require.NoError(t, c.Done()) + + select { + case got := <-collected: + sort.Ints(got) + require.Equal(t, []int{1, 2}, got) + case <-time.After(time.Second): + require.Fail(t, "Collect did not return after Done") + } +} + +func TestCollector_DoneConcurrent(t *testing.T) { + c := NewCollector[int]() + + require.NoError(t, c.Run(func(p Pusher[int]) error { + p.Push(1) + return nil + })) + + // Many concurrent Done calls must all return without panicking on a double close. + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + wg.Add(1) + go func() { + defer wg.Done() + require.NoError(t, c.Done()) + }() + } + wg.Wait() + + require.Equal(t, []int{1}, c.Collect()) +} + +var errTest = testError("test error") + +type testError string + +func (e testError) Error() string { return string(e) } diff --git a/pkg/util/k8sutil/lifecycle.go b/pkg/util/k8sutil/lifecycle.go index 268c7e6c1..ed07b66d8 100644 --- a/pkg/util/k8sutil/lifecycle.go +++ b/pkg/util/k8sutil/lifecycle.go @@ -1,7 +1,7 @@ // // DISCLAIMER // -// Copyright 2016-2025 ArangoDB GmbH, Cologne, Germany +// Copyright 2016-2026 ArangoDB GmbH, Cologne, Germany // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -112,6 +112,20 @@ func NewLifecycleWithBinary(exePath string, t string) (*core.Lifecycle, error) { return lifecycle, nil } +// NewCollectorPostStartHandler creates a postStart lifecycle handler which runs the collector. +func NewCollectorPostStartHandler() *core.LifecycleHandler { + return NewCollectorPostStartHandlerWithBinary(LifecycleBinary()) +} + +// NewCollectorPostStartHandlerWithBinary creates a postStart collector handler using a specific binary path. +func NewCollectorPostStartHandlerWithBinary(exePath string) *core.LifecycleHandler { + return &core.LifecycleHandler{ + Exec: &core.ExecAction{ + Command: append([]string{exePath}, "lifecycle", "postStart", "collector"), + }, + } +} + func AppendLifecycleEnv(in []core.EnvVar) []core.EnvVar { for _, e := range GetLifecycleEnv() { if !EnvExists(in, e.Name) { diff --git a/pkg/util/shutdown/shutdown.go b/pkg/util/shutdown/shutdown.go index 855f02bc3..adfc05527 100644 --- a/pkg/util/shutdown/shutdown.go +++ b/pkg/util/shutdown/shutdown.go @@ -1,7 +1,7 @@ // // DISCLAIMER // -// Copyright 2016-2024 ArangoDB GmbH, Cologne, Germany +// Copyright 2016-2026 ArangoDB GmbH, Cologne, Germany // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -25,6 +25,9 @@ import ( "os" "os/signal" "syscall" + "time" + + "k8s.io/apimachinery/pkg/util/uuid" ) func init() { @@ -43,8 +46,26 @@ func init() { var ( ctx context.Context stop context.CancelFunc + + // bootID is a unique identifier generated once per process boot. + bootID = string(uuid.NewUUID()) + + // bootTime is the timestamp captured once when the process boots. + bootTime = time.Now() ) +// BootID returns a unique identifier which is stable for the lifetime of the current process and +// changes on every (re)start. It can be used to correlate events emitted by a single boot. +func BootID() string { + return bootID +} + +// BootTime returns the timestamp captured once when the process booted. It is stable for the +// lifetime of the current process and pairs with BootID to identify a single boot. +func BootTime() time.Time { + return bootTime +} + func Context() context.Context { return ctx }