Add log stream creation and set retention in Helm Chart

helm/templates/_helpers_config_logstream.txt

@@ -0,0 +1,44 @@
+#!/bin/sh
+set -e ; # Have script exit in the event of a failed command.
+IFS=$' \t\r\n'
+
+createStream() {
+  STREAM=$1
+  echo;
+  echo \"Creating the log stream $STREAM\";
+
+  response=$(curl -sS --header 'Content-Type: application/json' -u "$P_USERNAME":"$P_PASSWORD" -w 'HTTPSTATUS:%{http_code}' --location --request PUT "http://{{ include "parseable.fullname" . }}.{{ .Release.Namespace }}:{{ $.Values.parseable.service.port }}/api/v1/logstream/$STREAM");
+
+  HTTP_BODY=$(echo $response | sed -e 's/HTTPSTATUS\:.*//g')
+  HTTP_STATUS=$(echo $response | tr -d '\n' | sed -e 's/.*HTTPSTATUS://')
+
+  echo \"API response:\"
+  echo \"HTTP status code: $HTTP_STATUS\"
+  echo \"HTTP response message: $HTTP_BODY\"
+}
+
+setRetention() {
+  STREAM=$1
+  ACTION=$2
+  DURATION=$3
+
+  echo;
+  echo \"Setting the retention for $STREAM\";
+
+  response=$(curl -sS --header 'Content-Type: application/json' -u "$P_USERNAME":"$P_PASSWORD" -w 'HTTPSTATUS:%{http_code}' --location --request PUT "http://{{ include "parseable.fullname" . }}.{{ .Release.Namespace }}:{{ $.Values.parseable.service.port }}/api/v1/logstream/$STREAM/retention" --data "[{\"description\":\"$ACTION logs after $DURATION\",\"action\":\"$ACTION\",\"duration\":\"$DURATION\"}]");
+
+  HTTP_BODY=$(echo $response | sed -e 's/HTTPSTATUS\:.*//g')
+  HTTP_STATUS=$(echo $response | tr -d '\n' | sed -e 's/.*HTTPSTATUS://')
+
+  echo \"API response:\"
+  echo \"HTTP status code: $HTTP_STATUS\"
+  echo \"HTTP response message: $HTTP_BODY\"
+}
+
+{{ if .Values.parseable.logstream }}
+# Create the logstream
+{{- range .Values.parseable.logstream }}
+createStream {{.name}}
+setRetention {{.name}} {{.retention.action}} {{.retention.duration}}
+{{- end }}
+{{- end }}

helm/templates/configmap.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ include "parseable.fullname" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "parseable.labels" . | nindent 4 }}
+data:
+  config-logstream: |-
+    {{- include (print $.Template.BasePath "/_helpers_config_logstream.txt") . | nindent 4 }}

helm/templates/deployment.yaml

@@ -2,6 +2,7 @@ apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: {{ include "parseable.fullname" . }}
+  namespace: {{ .Release.Namespace }}
   labels:
     {{- include "parseable.labels" . | nindent 4 }}
 spec:

helm/templates/logstream-job.yaml

@@ -0,0 +1,63 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: {{ include "parseable.fullname" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "parseable.labels" . | nindent 4 }}
+spec:
+  template:
+    metadata:
+      {{- with .Values.parseable.podAnnotations }}
+      annotations:
+        "helm.sh/hook": post-install,post-upgrade
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      labels:
+        {{- include "parseable.selectorLabels" . | nindent 8 }}
+    spec:
+      restartPolicy: OnFailure
+      securityContext:
+        {{- toYaml .Values.parseable.podSecurityContext | nindent 8 }}
+      volumes:
+        - name: parseable-init
+          projected:
+            sources:
+              - configMap:
+                  name: {{ include "parseable.fullname" . }}
+      containers:
+        - name: config-logstream
+          volumeMounts:
+            - name: parseable-init
+              mountPath: /config
+          securityContext:
+            {{- toYaml .Values.parseable.securityContext | nindent 12 }}
+          env:
+            {{- if .Values.parseable.local }}
+            {{- range $secret := .Values.parseable.localModeSecret }}
+            {{- range $key := $secret.keys }}
+            {{- $envPrefix := $secret.prefix | default "" | upper }}
+            {{- $envKey := $key | upper | replace "." "_" | replace "-" "_" }}
+            - name: {{ $envPrefix }}{{ $envKey }}
+              valueFrom:
+                secretKeyRef:
+                  name: {{ $secret.name }}
+                  key: {{ $key }}
+            {{- end }}
+            {{- end }}
+            {{- else}}
+            {{- range $secret := .Values.parseable.s3ModeSecret }}
+            {{- range $key := $secret.keys }}
+            {{- $envPrefix := $secret.prefix | default "" | upper }}
+            {{- $envKey := $key | upper | replace "." "_" | replace "-" "_" }}
+            - name: {{ $envPrefix }}{{ $envKey }}
+              valueFrom:
+                secretKeyRef:
+                  name: {{ $secret.name }}
+                  key: {{ $key }}
+            {{- end }}
+            {{- end }}
+            {{- end }}
+          image: curlimages/curl:8.00.0
+          command: [ "/bin/sh", "/config/config-logstream" ]
+  backoffLimit: 20

helm/values.yaml

@@ -3,9 +3,23 @@ parseable:
     repository: parseable/parseable
     tag: v0.4.3
     pullPolicy: Always
+  ## Set to true if you want to deploy Parseable in local mode (store logs 
+  ## on local mount point instead of S3 bucket)
   local: false
   # env:
-  #   RUST_LOG: info
+  #   RUST_LOG: error
+  # ## Enable to create and configure logstreams
+  # logstream:
+  #   - name: "vectordemo"
+  #     retention:
+  #       action: "delete"
+  #       duration: "30d"
+  #   - name: "fluentbitdemo"
+  #     retention:
+  #       action: "delete"
+  #       duration: "30d"
+  ## enable/disable persistence using PVC for the Data and Staging directories
+  ## Note that Data directory is needed only for local mode
   persistence:
     staging:
       enabled: false

server/src/handlers/http/ingest.rs

@@ -32,7 +32,7 @@ use crate::metadata::STREAM_INFO;
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 
 // Handler for POST /api/v1/ingest
-// ingests events by extacting stream name from header
+// ingests events by extracting stream name from header
 // creates if stream does not exist
 pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostError> {
     if let Some((_, stream_name)) = req