Merge branch 'main' into health

processor/signozlogspipelineprocessor/processor_test.go

@@ -3,6 +3,7 @@ package signozlogspipelineprocessor
 import (
 	"context"
 	"encoding/hex"
+	"fmt"
 	"testing"
 	"time"
 
@@ -305,6 +306,53 @@ func TestSignozLogPipelineWithRouterOp(t *testing.T) {
 	validateProcessorBehavior(t, confYaml, input, expectedOutput)
 }
 
+// logstransform in otel-collector-contrib doesn't support using severity text and number in route expressions
+func TestSeverityBasedRouteExpressions(t *testing.T) {
+	for _, routeExpr := range []string{
+		`severity_number == 9`,
+		`severity_text == "INFO"`,
+	} {
+		confYaml := fmt.Sprintf(`
+    operators:
+        - default: noop
+          id: router_signoz
+          routes:
+            - expr: %s
+              output: add-test-value
+          type: router
+        - id: add-test-value
+          on_error: send
+          type: add
+          field: attributes.test
+          value: test-value
+        - id: noop
+          type: noop
+    `, routeExpr)
+
+		// should process matching log
+		input := []plog.Logs{makePlogWithTopLevelFields(
+			t, "test log", map[string]any{},
+			map[string]any{"severity_text": "INFO", "severity_number": 9},
+		)}
+		expectedOutput := []plog.Logs{makePlogWithTopLevelFields(
+			t, "test log", map[string]any{"test": "test-value"},
+			map[string]any{"severity_text": "INFO", "severity_number": 9},
+		)}
+		validateProcessorBehavior(t, confYaml, input, expectedOutput)
+
+		// should ignore non-matching log
+		input = []plog.Logs{makePlogWithTopLevelFields(
+			t, "test log", map[string]any{},
+			map[string]any{"severity_text": "ERROR", "severity_number": 17},
+		)}
+		expectedOutput = []plog.Logs{makePlogWithTopLevelFields(
+			t, "test log", map[string]any{},
+			map[string]any{"severity_text": "ERROR", "severity_number": 17},
+		)}
+		validateProcessorBehavior(t, confYaml, input, expectedOutput)
+	}
+}
+
 func validateProcessorBehavior(
 	t *testing.T,
 	confYaml string,

processor/signozlogspipelineprocessor/stanza/operator/helper/expr.go

@@ -0,0 +1,38 @@
+// Mostly brought in as-is from otel-collector-contrib with minor changes
+// For example: includes severity_text and severity_number in GetExprEnv
+
+package signozstanzahelper
+
+import (
+	"os"
+	"sync"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+)
+
+var envPool = sync.Pool{
+	New: func() any {
+		return map[string]any{
+			"os_env_func": os.Getenv,
+		}
+	},
+}
+
+// GetExprEnv returns a map of key/value pairs that can be be used to evaluate an expression
+func GetExprEnv(e *entry.Entry) map[string]any {
+	env := envPool.Get().(map[string]any)
+	env["$"] = e.Body
+	env["body"] = e.Body
+	env["attributes"] = e.Attributes
+	env["resource"] = e.Resource
+	env["timestamp"] = e.Timestamp
+	env["severity_text"] = e.SeverityText
+	env["severity_number"] = int(e.Severity)
+
+	return env
+}
+
+// PutExprEnv adds a key/value pair that will can be used to evaluate an expression
+func PutExprEnv(e map[string]any) {
+	envPool.Put(e)
+}

processor/signozlogspipelineprocessor/stanza/operator/operators/add/transformer.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strings"
 
+	signozstanzahelper "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator/helper"
 	"github.com/expr-lang/expr/vm"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
@@ -33,8 +34,8 @@ func (t *Transformer) Transform(e *entry.Entry) error {
 		return e.Set(t.Field, t.Value)
 	}
 	if t.program != nil {
-		env := helper.GetExprEnv(e)
-		defer helper.PutExprEnv(env)
+		env := signozstanzahelper.GetExprEnv(e)
+		defer signozstanzahelper.PutExprEnv(env)
 
 		result, err := vm.Run(t.program, env)
 		if err != nil {

processor/signozlogspipelineprocessor/stanza/operator/operators/router/transformer.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 
+	signozstanzahelper "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator/helper"
 	"github.com/expr-lang/expr/vm"
 	"go.uber.org/zap"
 
@@ -35,8 +36,8 @@ func (t *Transformer) CanProcess() bool {
 
 // Process will route incoming entries based on matching expressions
 func (t *Transformer) Process(ctx context.Context, entry *entry.Entry) error {
-	env := helper.GetExprEnv(entry)
-	defer helper.PutExprEnv(env)
+	env := signozstanzahelper.GetExprEnv(entry)
+	defer signozstanzahelper.PutExprEnv(env)
 
 	for _, route := range t.routes {
 		matches, err := vm.Run(route.Expression, env)

receiver/clickhousesystemtablesreceiver/receiver.go

@@ -156,7 +156,12 @@ func (r *systemTablesReceiver) scrapeQueryLogIfReady(ctx context.Context) (
 	// For example, this can happen if this was the first successful scrape
 	// after several failed attempts and subsequent waits for r.ScrapeIntervalSeconds
 	nextScrapeMinServerTs := r.nextScrapeIntervalStartTs + r.scrapeIntervalSeconds + r.scrapeDelaySeconds
-	nextWaitSeconds := max(0, nextScrapeMinServerTs-serverTsNow)
+
+	nextWaitSeconds := uint32(0)
+	if nextScrapeMinServerTs > serverTsNow {
+		// Do the subtraction only if it will not lead to an overflow/wrap around
+		nextWaitSeconds = nextScrapeMinServerTs - serverTsNow
+	}
 
 	return nextWaitSeconds, nil
 }

receiver/clickhousesystemtablesreceiver/receiver_test.go

@@ -127,4 +127,14 @@ func TestReceiver(t *testing.T) {
 	et, exists := lr.Attributes().Get("event_time")
 	require.True(exists)
 	require.Equal(et.Str(), testQlEventTime.Format(time.RFC3339))
+
+	// should scrape again immediately if scrape is too far behind the server ts
+	// for example: this can happen if clickhouse goes down for some time
+	mockQuerrier.tsNow += 10 * testScrapeIntervalSeconds
+	testQl4 := makeTestQueryLog("host-4", time.Now(), "test query 4")
+	mockQuerrier.nextScrapeResult = []QueryLog{testQl4}
+
+	waitSeconds, err = testReceiver.scrapeQueryLogIfReady(context.Background())
+	require.Nil(err)
+	require.Equal(uint32(0), waitSeconds)
 }