logzio-fluentd-0.30.4

.github/workflows/logzio-fluentd-test.yaml

@@ -0,0 +1,79 @@
+name: Test `logzio-fluentd` chart
+
+on:
+  pull_request:
+    branches:
+      - master
+    paths:
+      - 'charts/fluentd/Chart.yaml'
+      - 'charts/fluentd/templates/**'
+      - 'charts/fluentd/values.yaml'
+
+jobs:
+  test-helm-chart:
+    name: Test Helm Chart on Kind
+    runs-on: ubuntu-latest
+    steps:
+      - name: Generate random id
+        id: random_id
+        run: echo "::set-output name=rand::$(echo $RANDOM)"
+
+      - name: Set ENV_ID
+        run: echo "ENV_ID=logs-test-run-${{ steps.random_id.outputs.rand }}" >> $GITHUB_ENV
+
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.20'
+
+      - name: Set up Helm
+        uses: azure/setup-helm@v4.2.0
+
+      - name: Set up kubectl
+        uses: azure/setup-kubectl@v4
+
+      - name: Install Kind
+        run: |
+          curl -Lo ./kind "https://kind.sigs.k8s.io/dl/v0.11.1/kind-Linux-amd64"
+          chmod +x ./kind
+          mv ./kind /usr/local/bin/kind
+
+      - name: Create Kind cluster
+        run: |
+          kind create cluster --name kind-${{ github.run_id }}
+          kubectl cluster-info
+
+      - name: Deploy Helm Chart
+        run: |
+          cd charts/fluentd
+          helm upgrade --install \
+          --set env_id=${{ env.ENV_ID }} \
+          --set secrets.logzioShippingToken=${{ secrets.LOGZIO_LOGS_TOKEN }} \
+          --set secrets.logzioListener=listener.logz.io \
+          logzio-fluentd .
+
+      - name: Run log generator
+        run: |
+          kubectl apply -f tests/resources/logsgen.yaml
+          kubectl rollout status deployment/log-generator --timeout=300s
+
+      - name: sleep
+        run: sleep 120
+
+      - name: Run Go Tests
+        env:
+          LOGZIO_LOGS_API_KEY: ${{ secrets.LOGZIO_LOGS_API_KEY }}
+        run: |
+          go get go.uber.org/zap
+          go test -v ./tests/fluentd_logs_e2e_test.go ./tests/common.go
+
+      - name: Cleanup Environment
+        run: |
+          helm uninstall logzio-fluentd
+
+      - name: Delete Kind cluster
+        if: always()
+        run: kind delete cluster --name kind-${{ github.run_id }}

charts/fluentd/Chart.yaml

@@ -4,7 +4,7 @@ description: A Helm chart for shipping Kubernetes logs via Fluentd.
 keywords:
   - logging
   - fluentd
-version: 0.30.3
+version: 0.30.4
 appVersion: 1.5.3
 maintainers:
   - name: Yotam loewenbach

charts/fluentd/README.md

@@ -302,6 +302,8 @@ If needed, the fluentd image can be changed to support windows server 2022 with
 
 
 ## Change log
+- **0.30.4**:
+   - Fix `nodeSelector` indentation
 - **0.30.3**:
   - Resolve `nodeSelector` bug
  - **0.30.2**:

charts/fluentd/templates/daemonset.yaml

@@ -42,7 +42,7 @@ spec:
       {{- end }}
       {{- with .Values.daemonset.nodeSelector }}
       nodeSelector:
-        {{- toYaml . | nindent 2 }}
+{{ toYaml . | indent 8 }}
       {{- end }}
       # Because the image's entrypoint requires to write on /fluentd/etc but we mount configmap there which is read-only,
       # this initContainers workaround or other is needed.
@@ -218,7 +218,7 @@ spec:
       {{- end }}
       {{- with .Values.windowsDaemonset.nodeSelector }}
       nodeSelector:
-        {{- toYaml . | nindent 2 }}
+{{ toYaml . | indent 8 }}
       {{- end }}
       # Because the image's entrypoint requires to write on /fluentd/etc but we mount configmap there which is read-only,
       # this initContainers workaround or other is needed.

tests/fluentd_logs_e2e_test.go

@@ -0,0 +1,112 @@
+package tests
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"go.uber.org/zap"
+	"io"
+	"net/http"
+	"os"
+	"reflect"
+	"testing"
+)
+
+type FluentdResponse struct {
+	Hits struct {
+		Total int `json:"total"`
+		Hits  []struct {
+			Source struct {
+				FluentTags string `json:"fluentd_tags"`
+				LogLevel   string `json:"log_level"`
+				Kubernetes struct {
+					ContainerImageTag  string `json:"container_image"`
+					ContainerImageName string `json:"container_image_id"`
+					ContainerName      string `json:"container_name"`
+					NamespaceName      string `json:"namespace_name"`
+					PodName            string `json:"pod_name"`
+					PodUID             string `json:"pod_id"`
+				} `json:"kubernetes"`
+			} `json:"_source"`
+		} `json:"hits"`
+	} `json:"hits"`
+}
+
+func TestLogzioFluentdLogs(t *testing.T) {
+	logsApiKey := os.Getenv("LOGZIO_LOGS_API_KEY")
+	if logsApiKey == "" {
+		t.Fatalf("LOGZIO_LOGS_API_KEY environment variable not set")
+	}
+
+	logResponse, err := fetchFluentdLogs(logsApiKey)
+	if err != nil {
+		t.Fatalf("Failed to fetch logs: %v", err)
+	}
+
+	if logResponse.Hits.Total == 0 {
+		t.Errorf("No logs found")
+	}
+
+	for _, hit := range logResponse.Hits.Hits {
+		kubernetes := hit.Source
+		if isAnyFieldEmpty(reflect.ValueOf(kubernetes)) {
+			logger.Error("Missing log fields", zap.Any("log", hit))
+			t.Errorf("Missing log fields")
+			break
+		}
+	}
+}
+func fetchFluentdLogs(logsApiKey string) (*FluentdResponse, error) {
+	url := fmt.Sprintf("%s/search", BaseLogzioApiUrl)
+	client := &http.Client{}
+	envID := os.Getenv("ENV_ID")
+	query := fmt.Sprintf("env_id:%s AND _exists_:fluentd_tags", envID)
+	formattedQuery := formatQuery(query)
+	logger.Info("sending api request", zap.String("url", url), zap.String("query", query))
+	req, err := http.NewRequest("POST", url, bytes.NewBufferString(formattedQuery))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-API-TOKEN", logsApiKey)
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	var logResponse FluentdResponse
+	err = json.Unmarshal(body, &logResponse)
+	if err != nil {
+		return nil, err
+	}
+
+	return &logResponse, nil
+}
+
+func isAnyFieldEmpty(logRes reflect.Value) bool {
+	for i := 0; i < logRes.NumField(); i++ {
+		switch logRes.Field(i).Kind() {
+		case reflect.String:
+			if logRes.Field(i).String() == "" {
+				return true
+			}
+		case reflect.Struct:
+			if isAnyFieldEmpty(logRes.Field(i)) {
+				return true
+			}
+		}
+	}
+	return false
+}