feat: release v0.0.6

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Chalkular Release Notes
 <!-- https://keepachangelog.com -->
+# [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

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

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: |-

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

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=

internal/reports/scheduler.go

@@ -18,18 +18,65 @@ 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 (
 	schedulerLabel = "chalk.ocular.crashoverride.run/scheduled-by"
 	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
 }

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)