Completed a working alpha for the operator

operator/api/v1alpha1/flowtest_types.go

@@ -32,16 +32,20 @@ type FlowTestSpec struct {
 
 // FlowTestStatus defines the observed state of FlowTest
 type FlowTestStatus struct {
-	FailedMatch  flowv1beta1.Match  `json:"failedMatch"`
-	FailedFilter flowv1beta1.Filter `json:"failedFilter"`
-	// +kubebuilder:validation:Enum=Created;Running;Completed
+	// +nullable
+	FailedMatches []flowv1beta1.Match `json:"failedMatches"`
+	// +nullable
+	FailedFilters []flowv1beta1.Filter `json:"failedFilters"`
+	// +kubebuilder:default:="Created"
+	// +kubebuilder:validation:Enum=Created;Running;Completed;Error
 	Status FlowStatus `json:"status"`
 }
 
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
-// +kubebuilder:printcolumn:JSONPath=".status.simulationPod.name",name="SimulationPod",type="string"
-// +kubebuilder:printcolumn:JSONPath=".status.simulationFlow.name",name="SimulationFlow",type="string"
+// +kubebuilder:printcolumn:JSONPath=".spec.referencePod.name",name="Reference Pod",type="string"
+// +kubebuilder:printcolumn:JSONPath=".spec.referenceFlow.name",name="Reference Flow",type="string"
+// +kubebuilder:printcolumn:JSONPath=".status.status",name="Status",type="string"
 
 // FlowTest is the Schema for the flowtests API
 type FlowTest struct {

operator/api/v1alpha1/reference_types.go

@@ -10,9 +10,8 @@ type ReferenceObject struct {
 type FlowStatus string
 
 const (
-	Created FlowStatus = "Created"
-	Running FlowStatus = "Running"
-	Skipped FlowStatus = "Skipped"
-	Failed  FlowStatus = "Failed"
-	Passed  FlowStatus = "Passed"
+	Created   FlowStatus = "Created"
+	Running   FlowStatus = "Running"
+	Completed FlowStatus = "Completed"
+	Error     FlowStatus = "Error"
 )

operator/config/manager/kustomization.yaml

@@ -5,6 +5,12 @@ generatorOptions:
   disableNameSuffixHash: true
 
 configMapGenerator:
-- name: manager-config
-  files:
+- files:
   - controller_manager_config.yaml
+  name: manager-config
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+images:
+- name: controller
+  newName: controller
+  newTag: latest

operator/config/rbac/role.yaml

@@ -6,6 +6,18 @@ metadata:
   creationTimestamp: null
   name: manager-role
 rules:
+- apiGroups:
+  - logging.banzaicloud.io
+  resources:
+  - clusterflows
+  - clusteroutputs
+  - flows
+  - outputs
+  verbs:
+  - create
+  - delete
+  - get
+  - list
 - apiGroups:
   - loggingplumber.isala.me
   resources:

operator/config/samples/loggingplumber_v1alpha1_flowtest.yaml

@@ -2,17 +2,22 @@ apiVersion: loggingplumber.isala.me/v1alpha1
 kind: FlowTest
 metadata:
   name: flowtest-sample
+  labels:
+    app.kubernetes.io/name: pod-simulation
+    app.kubernetes.io/managed-by: rancher-logging-explorer
+    app.kubernetes.io/created-by: logging-plumber
+    loggingplumber.isala.me/flowtest: flowtest-sample
 spec:
   referencePod:
     kind: Pod
-    name: test-pod
+    name: busybox-echo
     namespace: default
   referenceFlow:
-    kind: ClusterFlow
-    name: all-logs
+    kind: Flow
+    name: busybox-echo
     namespace: default
   sentMessages:
     - "[2021-06-10T11:50:06Z] @DEBUG Tam ipsae consuetudo infelix adtendi contexo mansuefecisti diutius re. 1373 ::0.403911"
     - "[2021-06-10T11:50:07Z] @WARNING Ne hi flagitantur alienam neglecta. 1374 ::0.474177"
     - "[2021-06-10T11:50:08Z] @INFO Amo ideoque die se at, caro aer, ad cor. 1375 ::0.263548"
-    - "[2021-06-10T11:50:09Z] @INFO Se contexo servis inpiis erogo, diligit ita significaret eosdem. 1376 ::0.405282"
+    - "[2021-06-10T11:50:09Z] @INFO Se contexo servis inpiis erogo, diligit ita significaret eosdem. 1376 ::0.405282"

operator/config/samples/simulations.yaml

@@ -0,0 +1,68 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: busybox-echo
+  labels:
+    loggingplumber.isala.me/test: simulations
+spec:
+  containers:
+    - name: busybox
+      image: busybox
+      command: ["sh", "-c", "x=1; while [ $x -gt -1 ]; do echo \"{'count': '$(( x++ ))', 'date': '$(date)'}\" && sleep 5; done"]
+---
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Flow
+metadata:
+  name: busybox-echo
+  labels:
+    loggingplumber.isala.me/test: simulations
+spec:
+  localOutputRefs:
+    - busybox-echo
+  match:
+    - select:
+        labels:
+          loggingplumber.isala.me/test: invalid
+    - select:
+        labels:
+          loggingplumber.isala.me/test: simulations
+  filters:
+    - record_modifier:
+        records:
+          - foo: "bar"
+    - grep:
+        regexp:
+          - key: first
+            pattern: /^5\d\d$/
+---
+apiVersion: logging.banzaicloud.io/v1beta1
+kind: Output
+metadata:
+  name: busybox-echo
+  labels:
+    loggingplumber.isala.me/test: simulations
+spec:
+  http:
+    endpoint: "http://logging-plumber-log-aggregator.default.svc/busybox-echo/"
+    buffer:
+      flush_interval: 10s
+      flush_mode: interval
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: logging-plumber-log-aggregator
+  labels:
+    loggingplumber.isala.me/test: simulations
+spec:
+  type: ClusterIP
+  ports:
+    - port: 80
+      targetPort: http
+      protocol: TCP
+      name: http
+  selector:
+    app.kubernetes.io/name: logging-plumber-log-aggregator
+    app.kubernetes.io/managed-by: rancher-logging-explorer
+    app.kubernetes.io/created-by: logging-plumber
+    loggingplumber.isala.me/component: log-aggregator

operator/controllers/cleanup.go

@@ -0,0 +1,74 @@
+package controllers
+
+import (
+	"context"
+	"fmt"
+	flowv1beta1 "github.com/banzaicloud/logging-operator/pkg/sdk/api/v1beta1"
+	v1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+func (r *FlowTestReconciler) cleanUpResources(ctx context.Context, flowTestName string) error {
+	logger := log.FromContext(ctx)
+
+	matchingLabels := &client.MatchingLabels{"loggingplumber.isala.me/flowtest": flowTestName}
+
+	var podList v1.PodList
+	if err := r.List(ctx, &podList, matchingLabels); client.IgnoreNotFound(err) != nil {
+		logger.Error(err, fmt.Sprintf("failed to get provisioned %s", podList.Kind))
+		return err
+	}
+
+	for _, resource := range podList.Items {
+		if err := r.Delete(ctx, &resource); client.IgnoreNotFound(err) != nil {
+			logger.Error(err, fmt.Sprintf("failed to delete a provisioned %s", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+			return err
+		}
+		logger.V(1).Info(fmt.Sprintf("%s deleted", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+	}
+
+	var configMapList v1.ConfigMapList
+	if err := r.List(ctx, &configMapList, matchingLabels); client.IgnoreNotFound(err) != nil {
+		logger.Error(err, fmt.Sprintf("failed to get provisioned %s", configMapList.Kind))
+		return err
+	}
+
+	for _, resource := range configMapList.Items {
+		if err := r.Delete(ctx, &resource); client.IgnoreNotFound(err) != nil {
+			logger.Error(err, fmt.Sprintf("failed to delete a provisioned %s", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+			return err
+		}
+		logger.V(1).Info(fmt.Sprintf("%s deleted", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+	}
+
+	var flows flowv1beta1.FlowList
+	if err := r.List(ctx, &flows, &client.MatchingLabels{"loggingplumber.isala.me/flowtest": flowTestName}); client.IgnoreNotFound(err) != nil {
+		logger.Error(err, fmt.Sprintf("failed to get provisioned %s", flows.Kind))
+		//return err
+	}
+
+	for _, resource := range flows.Items {
+		if err := r.Delete(ctx, &resource); client.IgnoreNotFound(err) != nil {
+			logger.Error(err, fmt.Sprintf("failed to delete a provisioned %s", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+			return err
+		}
+		logger.V(1).Info(fmt.Sprintf("%s deleted", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+	}
+
+	var outputs flowv1beta1.OutputList
+	if err := r.List(ctx, &outputs, &client.MatchingLabels{"loggingplumber.isala.me/flowtest": flowTestName}); client.IgnoreNotFound(err) != nil {
+		logger.Error(err, fmt.Sprintf("failed to get provisioned %s", outputs.Kind))
+		//return err
+	}
+
+	for _, resource := range outputs.Items {
+		if err := r.Delete(ctx, &resource); client.IgnoreNotFound(err) != nil {
+			logger.Error(err, fmt.Sprintf("failed to delete a provisioned %s", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+			return err
+		}
+		logger.V(1).Info(fmt.Sprintf("%s deleted", resource.Kind), "uuid", resource.GetUID(), "name", resource.GetName())
+	}
+
+	return nil
+}