diff --git a/CHANGELOG.md b/CHANGELOG.md index ee92e88..dbb2c7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Chalkular Release Notes +# [v0.0.6](https://github.com/crashappsec/chalkular/releases/tag/v0.0.6) - **June 26th, 2026** + +### Added + +- OpenMetrics for SQS and Scheduler + - SQS stats for messages read, errors and latency + - Scheduler stats for pipelines created and errors + +### Changed + +- upgrade ocular to v0.3.3 + # [v0.0.5](https://github.com/crashappsec/chalkular/releases/tag/v0.0.5) - **June 17th, 2026** ### Added diff --git a/cmd/uploader/main.go b/cmd/uploader/main.go index a71bd7d..ad8907e 100644 --- a/cmd/uploader/main.go +++ b/cmd/uploader/main.go @@ -72,6 +72,7 @@ func main() { l = l.WithValues( "workspace-id", workspaceID, "args", os.Args, "ingest-host", ingestHost, "s3-prefix", prefix, "s3-region", region, "s3-bucket", bucketName, + "metadata-id", metadataID, "action-id", actionID, ) var files []string diff --git a/config/crd/bases/chalk.ocular.crashoverride.run_chalkreportpolicies.yaml b/config/crd/bases/chalk.ocular.crashoverride.run_chalkreportpolicies.yaml index 1ed94f8..e58d461 100644 --- a/config/crd/bases/chalk.ocular.crashoverride.run_chalkreportpolicies.yaml +++ b/config/crd/bases/chalk.ocular.crashoverride.run_chalkreportpolicies.yaml @@ -243,6 +243,71 @@ spec: required: - name type: object + resources: + description: |- + Resources is the total amount of CPU and Memory resources required by all + containers for a pod. This is applied to both the scan pod and upload pod + (if the pipeline has one). + + This field enables fine-grained control over resource allocation for the + entire pod, allowing resource sharing among containers in a pod. + properties: + claims: + description: |- + Claims lists the names of resources, defined in spec.resourceClaims, + that are used by this container. + + This field depends on the + DynamicResourceAllocation feature gate. + + This field is immutable. It can only be set for containers. + items: + description: ResourceClaim references one entry in PodSpec.ResourceClaims. + properties: + name: + description: |- + Name must match the name of one entry in pod.spec.resourceClaims of + the Pod where this field is used. It makes that resource available + inside a container. + type: string + request: + description: |- + Request is the name chosen for a request in the referenced claim. + If empty, everything from the claim is made available, otherwise + only the result of this request. + type: string + required: + - name + type: object + type: array + x-kubernetes-list-map-keys: + - name + x-kubernetes-list-type: map + limits: + additionalProperties: + anyOf: + - type: integer + - type: string + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + description: |- + Limits describes the maximum amount of compute resources allowed. + More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ + type: object + requests: + additionalProperties: + anyOf: + - type: integer + - type: string + pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$ + x-kubernetes-int-or-string: true + description: |- + Requests describes the minimum amount of compute resources required. + If Requests is omitted for a container, it defaults to Limits if that is explicitly specified, + otherwise to an implementation-defined value. Requests cannot exceed Limits. + More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ + type: object + type: object runtimeClassName: description: |- RuntimeClassName is the name of the RuntimeClass that will be used to run the scan and upload pods. @@ -288,7 +353,7 @@ spec: after the specified number of seconds have passed since the pipeline was created, regardless of its state. format: int32 - minimum: 0 + minimum: 1 type: integer uploadServiceAccountName: description: |- diff --git a/go.mod b/go.mod index 139fe6e..612fd19 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/s3 v1.100.0 github.com/aws/aws-sdk-go-v2/service/sqs v1.42.21 github.com/awslabs/amazon-ecr-credential-helper/ecr-login v0.12.0 - github.com/crashappsec/ocular v0.3.1 + github.com/crashappsec/ocular v0.3.3 github.com/gin-gonic/gin v1.11.0 github.com/google/cel-go v0.28.1 github.com/google/go-containerregistry v0.21.5 diff --git a/go.sum b/go.sum index 4066d84..1cc518a 100644 --- a/go.sum +++ b/go.sum @@ -85,6 +85,10 @@ github.com/containerd/stargz-snapshotter/estargz v0.18.2/go.mod h1:XyVU5tcJ3PRpk github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/crashappsec/ocular v0.3.1 h1:lqq1ZuA0tzDoj8ginv6Bh6MVdzRHXm2ZL5od3s/8sj8= github.com/crashappsec/ocular v0.3.1/go.mod h1:TXYwo5LoRZkYH4dKbUmw/5TrkKER+chbsR8rGml5fBA= +github.com/crashappsec/ocular v0.3.3-beta.2 h1:N1OaZeXBaVxbiCDossgagWnvUFGJqdMCk6oAE4pw/QQ= +github.com/crashappsec/ocular v0.3.3-beta.2/go.mod h1:lg7VzhAjqT9NeRQMWhrtZsBdKgfhJ0snPkNXyNIpNC8= +github.com/crashappsec/ocular v0.3.3 h1:sm4SqHznQZirvnflj8ppLjaMLOiQI5NK01OdlEA2qVQ= +github.com/crashappsec/ocular v0.3.3/go.mod h1:lg7VzhAjqT9NeRQMWhrtZsBdKgfhJ0snPkNXyNIpNC8= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= diff --git a/internal/reports/scheduler.go b/internal/reports/scheduler.go index 3f1a488..e64e188 100644 --- a/internal/reports/scheduler.go +++ b/internal/reports/scheduler.go @@ -18,11 +18,13 @@ import ( "github.com/crashappsec/chalkular/api/v1beta1/chalk" "github.com/crashappsec/chalkular/internal/policy" ocularv1beta1 "github.com/crashappsec/ocular/api/v1beta1" + "github.com/prometheus/client_golang/prometheus" corev1 "k8s.io/api/core/v1" "k8s.io/client-go/tools/events" "sigs.k8s.io/controller-runtime/pkg/client" logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/metrics" ) const ( @@ -30,6 +32,51 @@ const ( schedulerValue = "chalkular-controller" ) +var ( + schedulerPipelinesCreated = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "scheduler_pipelines_created", + Help: "Total number of pipelines created", + }, + []string{"profile", "policy", "namespace"}, + ) + schedulerReportErrors = prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "scheduler_reports_errors", + Help: "Total number of pipelines created", + }, + ) + schedulerReportsRecieved = prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "scheduler_reports_received", + Help: "Total number of pipelines created", + }, + ) + schedulerEventsRecieved = prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "scheduler_events_received", + Help: "Total number of pipelines created", + }, + ) + schedulerEventProcessingDurationSeconds = prometheus.NewHistogram( + prometheus.HistogramOpts{ + Name: "scheduler_event_processing_duration_seconds", + Help: "Duration in seconds of processing per event", + }, + ) +) + +func init() { + // Register custom metrics with the global prometheus registry + metrics.Registry.MustRegister( + schedulerEventProcessingDurationSeconds, + schedulerEventsRecieved, + schedulerPipelinesCreated, + schedulerReportErrors, + schedulerReportsRecieved, + ) +} + type Scheduler struct { eventBus eventBus rejectPipelineThreshold int @@ -83,92 +130,98 @@ func (s *Scheduler) Start(ctx context.Context) error { select { case <-ctx.Done(): return ctx.Err() - case event := <-s.eventBus: + case e := <-s.eventBus: l.Info("chalk reports received, scheduling") - reports := event.Reports - - if s.rejectPipelineThreshold > 0 { - active, err := s.countActivePipelines(ctx) - if err != nil { - event.Result <- fmt.Errorf("unable to list active pipelines: %w", err) - close(event.Result) - continue - } - if active >= s.rejectPipelineThreshold { - l.Info("rejecting reports, pipeline threshold hit") - event.Result <- fmt.Errorf("%w: currently %d active which exceeds threshold of %d, rejecting event", - ErrPipelineThreshold, active, s.rejectPipelineThreshold) - close(event.Result) - continue - } + schedulerEventsRecieved.Inc() + schedulerReportsRecieved.Add(float64(len(e.Reports))) + start := time.Now() + err := s.processReports(ctx, e.Reports) + e.Result <- err + close(e.Result) + duration := time.Since(start) + schedulerEventProcessingDurationSeconds.Observe(duration.Seconds()) + if err != nil { + schedulerReportErrors.Add(1) } + time.Sleep(time.Second) + } + } +} - policies := &chalkularv1beta1.ChalkReportPolicyList{} - if err := s.mgrClient.List(ctx, policies); err != nil { - event.Result <- fmt.Errorf("unable to list chalk report policies: %w", err) - close(event.Result) - } +func (s *Scheduler) processReports(ctx context.Context, reports []chalk.Report) error { + l := logf.FromContext(ctx) + l.Info("chalk reports received, scheduling") - // group generated pipelines by report + policy - // so that we can write events to policies if templated pipeline - // fails to be created - var generatedPipelines []policyGeneratedPipelines - for _, report := range reports { - actionID, exist := report[chalk.KeyActionID] - actionIDStr, valid := actionID.(string) - if !exist || !valid { - l.Error(fmt.Errorf("missing or invalid key \"%s\" found in report", chalk.KeyActionID), "action ID string was not found for report") - event.Result <- fmt.Errorf("invalid chalk report, missing or invalid key %s found", chalk.KeyActionID) - close(event.Result) - continue - } - reportL := l.WithValues("action-id", actionID) - reportCtx := logf.IntoContext(ctx, reportL) - - generated := s.createPipelinesForReport(reportCtx, policies.Items, actionIDStr, report) - generatedPipelines = append(generatedPipelines, generated...) - } + if s.rejectPipelineThreshold > 0 { + active, err := s.countActivePipelines(ctx) + if err != nil { + return fmt.Errorf("unable to list active pipelines: %w", err) + } + if active >= s.rejectPipelineThreshold { + l.Info("rejecting reports, pipeline threshold hit") + return fmt.Errorf("%w: currently %d active which exceeds threshold of %d, rejecting event", + ErrPipelineThreshold, active, s.rejectPipelineThreshold) + } + } - // this is separate incase we fail to process one report, - // we reject before pipelines are created in order to allow - // the message to be requeued. If pipelines fail to be created - // the errors will be logged to the poilicies - // events and is considered an error with the policy, - // not the report. - var createdPipelines []*ocularv1beta1.Pipeline - for _, g := range generatedPipelines { - var policyPipelines []*ocularv1beta1.Pipeline - for i, pipeline := range g.pipelines { - err := s.mgrClient.Create(ctx, pipeline) - if err != nil { - l.Error(err, "unable to create pipeline for policy", - "pipeline", pipeline.Name, "namespace", g.policy.Namespace, "policy", g.policy.Name) - s.recorder.Eventf(&g.policy, nil, - corev1.EventTypeWarning, - "FailedToCreatePipeline", - "CreatePipelineFromReport", - "failed to generate pipeline (%d/%d) for report '%s': %s", i, len(g.pipelines), g.actionID, err) - } else { - policyPipelines = append(policyPipelines, pipeline) - } - } - if len(policyPipelines) > 0 { - s.recorder.Eventf(&g.policy, nil, - corev1.EventTypeNormal, - "PipelinesCreated", - "CreatePipelineFromReport", - "report '%s' created %d pipeline", g.actionID, len(createdPipelines)) - createdPipelines = append(createdPipelines, policyPipelines...) - } + policies := &chalkularv1beta1.ChalkReportPolicyList{} + if err := s.mgrClient.List(ctx, policies); err != nil { + return fmt.Errorf("unable to list chalk report policies: %w", err) + } - } + // group generated pipelines by report + policy + // so that we can write events to policies if templated pipeline + // fails to be created + var generatedPipelines []policyGeneratedPipelines + for _, report := range reports { + actionID, exist := report[chalk.KeyActionID] + actionIDStr, valid := actionID.(string) + if !exist || !valid { + l.Error(fmt.Errorf("missing or invalid key \"%s\" found in report", chalk.KeyActionID), "action ID string was not found for report") + return fmt.Errorf("invalid chalk report, missing or invalid key %s found", chalk.KeyActionID) + } + reportL := l.WithValues("action-id", actionID) + reportCtx := logf.IntoContext(ctx, reportL) - l.Info(fmt.Sprintf("created %d pipelines", len(createdPipelines)), "pipelines", len(createdPipelines)) - event.Result <- nil - close(event.Result) - // if we get here lets rest a second to not - // overwhelm the api server - time.Sleep(time.Second) + generated := s.createPipelinesForReport(reportCtx, policies.Items, actionIDStr, report) + generatedPipelines = append(generatedPipelines, generated...) + } + + // this is separate incase we fail to process one report, + // we reject before pipelines are created in order to allow + // the message to be requeued. If pipelines fail to be created + // the errors will be logged to the poilicies + // events and is considered an error with the policy, + // not the report. + var createdPipelines []*ocularv1beta1.Pipeline + for _, g := range generatedPipelines { + var policyPipelines []*ocularv1beta1.Pipeline + for i, pipeline := range g.pipelines { + err := s.mgrClient.Create(ctx, pipeline) + if err != nil { + l.Error(err, "unable to create pipeline for policy", + "pipeline", pipeline.Name, "namespace", g.policy.Namespace, "policy", g.policy.Name) + s.recorder.Eventf(&g.policy, nil, + corev1.EventTypeWarning, + "FailedToCreatePipeline", + "CreatePipelineFromReport", + "failed to generate pipeline (%d/%d) for report '%s': %s", i, len(g.pipelines), g.actionID, err) + } else { + schedulerPipelinesCreated.With(prometheus.Labels{"profile": pipeline.Spec.ProfileRef.Name, "policy": g.policy.Name, "namespace": pipeline.Namespace}) + policyPipelines = append(policyPipelines, pipeline) + } } + if len(policyPipelines) > 0 { + s.recorder.Eventf(&g.policy, nil, + corev1.EventTypeNormal, + "PipelinesCreated", + "CreatePipelineFromReport", + "report '%s' created %d pipeline", g.actionID, len(createdPipelines)) + createdPipelines = append(createdPipelines, policyPipelines...) + } + } + + l.Info(fmt.Sprintf("created %d pipelines", len(createdPipelines)), "pipelines", len(createdPipelines)) + return nil } diff --git a/internal/reports/sqs/listener.go b/internal/reports/sqs/listener.go index cf0e2b6..60dc7d8 100644 --- a/internal/reports/sqs/listener.go +++ b/internal/reports/sqs/listener.go @@ -18,6 +18,7 @@ import ( "github.com/crashappsec/chalkular/internal/reports" "github.com/prometheus/client_golang/prometheus" logf "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/metrics" ) var ( @@ -54,6 +55,17 @@ var ( ) ) +func init() { + // Register custom metrics with the global prometheus registry + metrics.Registry.MustRegister( + sqsMessageProcessingDurationSeconds, + sqsMessagesDeletedTotal, + sqsMessagesProcessedTotal, + sqsMessagesReceivedTotal, + sqsReceiveErrorsTotal, + ) +} + type SQSClientAPI interface { ReceiveMessage(ctx context.Context, params *sqs.ReceiveMessageInput, optFns ...func(*sqs.Options)) (*sqs.ReceiveMessageOutput, error)