Merge branch 'main' into garbage-collector-cronjob

Author: s-vitaliy

docs/user_scenarios.md

@@ -23,7 +23,28 @@ The backfill request will be picked up by the operator, which will create a Kube
 
 If the stream **is suspended**, you need to create backfill request first and then unsuspend the stream.
 
+Example backfill request:
+```yaml
+apiVersion: streaming.sneaksanddata.com/v1
+kind: BackfillRequest
+metadata:
+  name: my-backfill-request
+  namespace: arcane-stream-mock
+spec:
+  streamClass: arcane-stream-mock
+  streamId: my-stream
+```
+
 ## My stream has failed and I want to restart it
 If your stream has failed, you can set `spec.suspended` to `true` to stop the stream.
 To avoid data loss, you may create a backfill request that fills in any gaps occurred during the failure.
 After that, you can set `spec.suspended` to `false` to restart the stream.
+
+## I deleted the pod and my stream transitioned to failed state, how do I avoid that in the future?
+Arcane streaming is built on top of Kubernetes Jobs. By default, when a pod is deleted manually or due to node eviction,
+all containers in the pod receive a SIGTERM signal and have a grace period to shut down gracefully **with exit code 0**.
+If the containers do not shut down within the grace period, the pod is forcefully terminated. If the pod is terminated
+with a non-zero exit code, Kubernetes counts this pod **as failed** and the job may transition to a failed state.
+It's a responsibility of the user and/or the plugin developer to ensure that the streaming job handles termination signals
+gracefully and exits with code 0 or the exit code that returned by the plugin executable on termination is
+[added to the job's podFailurePolicy](https://kubernetes.io/docs/tasks/job/pod-failure-policy/).

justfile

@@ -27,7 +27,7 @@ mock-stream-plugin:
         --namespace default \
         --set jobTemplateSettings.podFailurePolicySettings.retryOnExitCodes="{120,121}" \
         --set jobTemplateSettings.backoffLimit=1 \
-        --version v1.0.4
+        --version v1.0.5
 
 manifests:
     kubectl apply -f integration_tests/manifests/*.yaml

performance_tests/manifests/readme.md

@@ -0,0 +1,137 @@
+# Performance Testing Resources for Arcane Operator
+
+## Overview
+
+This directory contains Kubernetes manifests for performance testing the Arcane operator. The resources enable automated testing of streaming workloads at scale.
+
+## Prerequisites
+
+- Kubernetes cluster with Arcane operator and `arcane-stream-mock` installed
+- `kubectl` configured to access the target cluster
+- Namespace `arcane-stream-mock` created
+
+## Resources
+
+### 1. test_secret.yaml
+A Kubernetes Secret containing test credentials and configuration data.
+
+**Purpose**: Provides ability to test secret management and access by streaming jobs.
+
+**Apply**:
+```bash
+kubectl apply -f test_secret.yaml
+```
+
+### 2. stream_template.yaml
+A `TestStreamDefinition` resource that defines mock streaming workloads.
+
+**Purpose**: Creates test streams for performance evaluation.
+
+**Key Parameters**:
+- `runDuration`: Duration for each test run (default: 360s)
+- `suspended`: Set to `false` to start streams in backfill mode
+- `shouldFail`: Set to `true` to test failure scenarios
+
+**Apply**:
+```bash
+kubectl apply -f stream_template.yaml
+```
+
+## Performance Testing Workflow
+
+### Step 1: Deploy Test Secret
+```bash
+kubectl apply -f test_secret.yaml
+```
+
+Verify the secret is created:
+```bash
+kubectl get secret test-secret -n arcane-stream-mock
+```
+
+### Step 2: Create StreamingJobTemplate
+Ensure the referenced `StreamingJobTemplate` (`arcane-stream-mock-standard-job`) exists in the `arcane-stream-mock` namespace before applying the stream template.
+
+### Step 3: Deploy Test Streams
+```bash
+for i in {1..2000}; do kubectl create -f stream_template.yaml; done
+```
+
+This creates a two thousand test streams with a generated name (e.g., `mock-stream-abc123`).
+
+### Step 4: Activate Test Streams
+By default, streams are created in suspended state. To start testing:
+
+Assuming you want to activate 300 streams at a time in parallel with 4 concurrent processes:
+```bash
+kubectl get teststreamdefinition -n arcane-stream-mock --no-headers \
+  | grep Suspended \
+  | head -n 300 \
+  | awk '{print $1}' \
+  | xargs --max-procs=4 \
+  | kubectl patch teststreamdefinition <stream-name> -n arcane-stream-mock -p '{"spec":{"suspended":false}}' --type=merge
+```
+
+### Step 5: Monitor Performance
+Use the Prometeus metrics exposed by the Arcane operator to monitor performance and resource usage during testing.
+
+## Scaling Tests
+
+## Customizing Tests
+
+### Modify Run Duration
+Edit `stream_template.yaml` and change `runDuration`:
+```yaml
+spec:
+  runDuration: 600s  # 10 minutes
+```
+
+### Test Failure Scenarios
+```yaml
+spec:
+  shouldFail: true
+```
+
+## Cleanup
+
+### Remove All Test Streams
+```bash
+kubectl delete teststreamdefinition -n arcane-stream-mock --all
+```
+
+### Remove Test Secret
+```bash
+kubectl delete -f test_secret.yaml
+```
+
+## Troubleshooting
+
+### Stream Not Starting
+- Check if `suspended` is set to `false`
+- Verify `StreamingJobTemplate` reference exists
+- Check operator logs: `kubectl logs -n <operator-namespace> -l app=arcane-operator`
+
+### Jobs Failing
+- Verify test secret is properly configured
+- Check job logs: `kubectl logs -n arcane-stream-mock <job-pod-name>`
+- Ensure sufficient cluster resources (CPU, memory)
+- Check if `shouldFail` is set to `true` for intentional failures
+
+## Metrics Collection
+
+Monitor operator performance metrics:
+```bash
+# CPU/Memory usage
+kubectl top pods -n <operator-namespace>
+
+# Stream processing metrics
+kubectl describe teststreamdefinition -n arcane-stream-mock
+```
+
+## Best Practices
+
+1. **Incremental Load Testing**: Start with 1-2 streams, gradually increase
+2. **Resource Limits**: Monitor cluster resources during tests
+3. **Cleanup**: Always clean up test resources after performance testing
+4. **Isolation**: Use dedicated namespace for performance tests
+5. **Baseline**: Establish baseline metrics before scaling tests

performance_tests/manifests/stream_template.yaml

@@ -0,0 +1,23 @@
+apiVersion: streaming.sneaksanddata.com/v1
+kind: TestStreamDefinition
+metadata:
+  generateName: mock-stream-
+  namespace: arcane-stream-mock
+spec:
+  backfillJobTemplateRef:
+    apiVersion: streaming.sneaksanddata.com/v1
+    kind: StreamingJobTemplate
+    name: arcane-stream-mock-standard-job
+    namespace: arcane-stream-mock
+  destination: mock-destination
+  jobTemplateRef:
+    apiVersion: streaming.sneaksanddata.com/v1
+    kind: StreamingJobTemplate
+    name: arcane-stream-mock-standard-job
+    namespace: arcane-stream-mock
+  runDuration: 360s
+  shouldFail: false
+  source: mock-source
+  suspended: true
+  testSecretRef:
+    name: test-secret

performance_tests/manifests/test_secret.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: test-secret
+  namespace: arcane-stream-mock
+data:
+  TEST_SECRET: c2VjcmV0IGRhdGE=
+type: Opaque
+
+

services/controllers/stream/stream_reconciler.go

@@ -497,7 +497,7 @@ func (s *streamReconciler) getLogger(_ context.Context, request types.Namespaced
 func (s *streamReconciler) updateStreamPhase(ctx context.Context, definition Definition, backfillRequest *v1.BackfillRequest, next Phase, eventFunc controllers.EventFunc) (reconcile.Result, error) {
 	logger := s.getLogger(ctx, definition.NamespacedName())
 	if definition.GetPhase() == next { // coverage-ignore
-		logger.V(0).Info("Stream phase is already set to", "phase", definition.GetPhase())
+		logger.V(0).Info("Stream phase is already set", "phase", definition.GetPhase())
 		return reconcile.Result{}, nil
 	}
 	logger.V(0).Info("updating Stream status", "from", definition.GetPhase(), "to", next)

services/controllers/stream/streaming_job.go

@@ -18,11 +18,21 @@ func (j StreamingJob) CurrentConfiguration() (string, error) { // coverage-ignor
 }
 
 func (j StreamingJob) IsCompleted() bool { // coverage-ignore (trivial)
-	return j.Status.Succeeded > 0
+	for _, condition := range j.Status.Conditions {
+		if condition.Type == v1.JobComplete && condition.Status == "True" {
+			return true
+		}
+	}
+	return false
 }
 
 func (j StreamingJob) IsFailed() bool { // coverage-ignore (trivial)
-	return j.Status.Failed > 0 && j.Status.Failed >= *j.Spec.BackoffLimit
+	for _, condition := range j.Status.Conditions {
+		if condition.Type == v1.JobFailed && condition.Status == "True" {
+			return true
+		}
+	}
+	return false
 }
 
 func (j StreamingJob) ToV1Job() *v1.Job { // coverage-ignore (trivial)