feat(metering): add metering for logs (#493)

#### Features

- Add a pkg `metering` with interface `metering.Meter` for Size and Count usage calculation
- Add a pkg `pdatagen` for generating a wide variety of telemetry data for testing

pkg/metering/json.go

@@ -0,0 +1,27 @@
+package metering
+
+import (
+	"encoding/json"
+
+	"go.uber.org/zap"
+)
+
+type jsonSizer struct {
+	Logger *zap.Logger
+}
+
+func NewJSONSizer(logger *zap.Logger) *jsonSizer {
+	return &jsonSizer{
+		Logger: logger,
+	}
+}
+
+func (sizer *jsonSizer) SizeOfMapStringAny(input map[string]any) int {
+	bytes, err := json.Marshal(input)
+	if err != nil {
+		sizer.Logger.Error("cannot marshal object, setting size to 0", zap.Any("obj", input))
+		return 0
+	}
+
+	return len(bytes)
+}

pkg/metering/meter.go

@@ -0,0 +1,37 @@
+package metering
+
+import (
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+// Meter is an interface that receives telemetry data and
+// calculates billable metrics.
+type Meter[T ptrace.Traces | pmetric.Metrics | plog.Logs] interface {
+	// Calculates size of the telemetry data in bytes.
+	Size(T) int
+	// Calculates count of the telemetry data.
+	Count(T) int
+}
+
+// Sizer is an interface that calculates the size of different
+// data structures
+type Sizer interface {
+	SizeOfMapStringAny(map[string]any) int
+}
+
+// Calculates billable metrics for logs.
+type Logs interface {
+	Meter[plog.Logs]
+}
+
+// Calculates billable metrics for traces.
+type Traces interface {
+	Meter[ptrace.Traces]
+}
+
+// Calculates billable metrics for metrics.
+type Metrics interface {
+	Meter[pmetric.Metrics]
+}

pkg/metering/v1/logs.go

@@ -0,0 +1,45 @@
+package v1
+
+import (
+	"github.com/SigNoz/signoz-otel-collector/pkg/metering"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+)
+
+type logs struct {
+	Logger *zap.Logger
+	Sizer  metering.Sizer
+}
+
+func NewLogs(logger *zap.Logger) metering.Logs {
+	return &logs{
+		Logger: logger,
+		Sizer:  metering.NewJSONSizer(logger),
+	}
+}
+
+func (meter *logs) Size(ld plog.Logs) int {
+	total := 0
+
+	for i := 0; i < ld.ResourceLogs().Len(); i++ {
+		resourceLog := ld.ResourceLogs().At(i)
+		resourceAttributesSize := meter.Sizer.SizeOfMapStringAny(resourceLog.Resource().Attributes().AsRaw())
+
+		for j := 0; j < resourceLog.ScopeLogs().Len(); j++ {
+			scopeLogs := resourceLog.ScopeLogs().At(j)
+
+			for k := 0; k < scopeLogs.LogRecords().Len(); k++ {
+				logRecord := scopeLogs.LogRecords().At(k)
+				total += resourceAttributesSize +
+					meter.Sizer.SizeOfMapStringAny(logRecord.Attributes().AsRaw()) +
+					len([]byte(logRecord.Body().AsString()))
+			}
+
+		}
+	}
+
+	return total
+}
+func (*logs) Count(ld plog.Logs) int {
+	return ld.LogRecordCount()
+}

pkg/metering/v1/logs_test.go

@@ -0,0 +1,65 @@
+package v1
+
+import (
+	"testing"
+
+	"github.com/SigNoz/signoz-otel-collector/pkg/pdatagen/plogsgen"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+)
+
+func TestLogsSize(t *testing.T) {
+	logs := plogsgen.Generate(
+		plogsgen.WithLogRecordCount(10),
+		plogsgen.WithResourceAttributeCount(8),
+		// 100 bytes
+		plogsgen.WithBody("Lorem ipsum dolor sit amet consectetur adipiscing elit, enim suscipit nullam aenean mattis senectus."),
+		// 20 bytes
+		plogsgen.WithResourceAttributeStringValue("Lorem ipsum euismod."),
+	)
+
+	meter := NewLogs(zap.NewNop())
+	size := meter.Size(logs)
+	// 8 * [ 10(key) + 20(value) + 5("":"") ] + 2({}) + 7(,)
+	assert.Equal(t, 10*(8*(10+20+5)+7+2+2+100), size)
+}
+
+func benchmarkLogsSize(b *testing.B, expectedSize int, options ...plogsgen.GenerationOption) {
+	b.Helper()
+
+	logs := plogsgen.Generate(options...)
+	meter := NewLogs(zap.NewNop())
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for n := 0; n < b.N; n++ {
+		size := meter.Size(logs)
+		assert.Equal(b, expectedSize, size)
+	}
+}
+
+func BenchmarkLogsSize_20000_20(b *testing.B) {
+	benchmarkLogsSize(
+		b,
+		16660000,
+		plogsgen.WithLogRecordCount(20000),
+		plogsgen.WithResourceAttributeCount(20),
+		// 100 bytes
+		plogsgen.WithBody("Lorem ipsum dolor sit amet consectetur adipiscing elit, enim suscipit nullam aenean mattis senectus."),
+		// 20 bytes
+		plogsgen.WithResourceAttributeStringValue("Lorem ipsum euismod."),
+	)
+}
+
+func BenchmarkLogsSize_100000_20(b *testing.B) {
+	benchmarkLogsSize(
+		b,
+		83300000,
+		plogsgen.WithLogRecordCount(100000),
+		plogsgen.WithResourceAttributeCount(20),
+		// 100 bytes
+		plogsgen.WithBody("Lorem ipsum dolor sit amet consectetur adipiscing elit, enim suscipit nullam aenean mattis senectus."),
+		// 20 bytes
+		plogsgen.WithResourceAttributeStringValue("Lorem ipsum euismod."),
+	)
+}

pkg/pdatagen/plogsgen/logs.go

@@ -0,0 +1,42 @@
+package plogsgen
+
+import (
+	"strconv"
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+func Generate(opts ...GenerationOption) plog.Logs {
+	generationOpts := generationOptions{
+		logRecordCount:               1,
+		resourceAttributeCount:       1,
+		body:                         "This is a test log record",
+		resourceAttributeStringValue: "resource",
+	}
+
+	for _, opt := range opts {
+		opt(&generationOpts)
+	}
+
+	endTime := pcommon.NewTimestampFromTime(time.Now())
+	logs := plog.NewLogs()
+	resourceLog := logs.ResourceLogs().AppendEmpty()
+	for i := 0; i < generationOpts.resourceAttributeCount; i++ {
+		suffix := strconv.Itoa(i)
+		// Do not change the key name format in resource attributes below.
+		resourceLog.Resource().Attributes().PutStr("resource."+suffix, generationOpts.resourceAttributeStringValue)
+	}
+
+	scopeLogs := resourceLog.ScopeLogs().AppendEmpty()
+	scopeLogs.LogRecords().EnsureCapacity(generationOpts.logRecordCount)
+	for i := 0; i < generationOpts.logRecordCount; i++ {
+		logRecord := scopeLogs.LogRecords().AppendEmpty()
+		logRecord.SetTimestamp(endTime)
+		logRecord.SetObservedTimestamp(endTime)
+		logRecord.Body().SetStr(generationOpts.body)
+	}
+
+	return logs
+}

pkg/pdatagen/plogsgen/options.go

@@ -0,0 +1,34 @@
+package plogsgen
+
+type generationOptions struct {
+	logRecordCount               int
+	resourceAttributeCount       int
+	body                         string
+	resourceAttributeStringValue string
+}
+
+type GenerationOption func(*generationOptions)
+
+func WithLogRecordCount(i int) GenerationOption {
+	return func(o *generationOptions) {
+		o.logRecordCount = i
+	}
+}
+
+func WithResourceAttributeCount(i int) GenerationOption {
+	return func(o *generationOptions) {
+		o.resourceAttributeCount = i
+	}
+}
+
+func WithBody(s string) GenerationOption {
+	return func(o *generationOptions) {
+		o.body = s
+	}
+}
+
+func WithResourceAttributeStringValue(s string) GenerationOption {
+	return func(o *generationOptions) {
+		o.resourceAttributeStringValue = s
+	}
+}