Trace Events Exporting As Logs

Dockerfile

@@ -5,7 +5,7 @@ ARG TARGETARCH
 
 RUN apk update && apk add --no-cache git bash ca-certificates && update-ca-certificates
 
-ARG BUILD_ID="17.0.1"
+ARG BUILD_ID="17.0.2"
 
 WORKDIR /app
 

build/kubernetes/helm/opsramp-tracing-proxy/Chart.yaml

@@ -7,10 +7,10 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 17.0.1
+version: 17.0.2
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: "17.0.1"
+appVersion: "17.0.2"

build/kubernetes/helm/opsramp-tracing-proxy/values.yaml

@@ -127,6 +127,12 @@ config:
   # Duration is difference between end and start time of a span
   Threshold: 500
 
+  # Url to push events as logs
+  LogsEndpoint: "<OPSRAMP_LOGS_API>"
+
+  # If true trace proxy will send span events as logs
+  SendEvents: true
+
   # EnvironmentCacheTTL is the amount of time a cache entry will live that associates
   # an API key with an environment name.
   # Cache misses lookup the environment name using OpsRampAPI config value.

build/kubernetes/yaml/k8s-config-cm.yaml

@@ -108,6 +108,12 @@ data:
     # Duration is difference between end and start time of a span
     Threshold: 500
 
+    # Url to push events as logs
+    LogsEndpoint: "<OPSRAMP_LOGS_API>"
+
+    # If true trace proxy will send span events as logs
+    SendEvents: true
+
     # EnvironmentCacheTTL is the amount of time a cache entry will live that associates
     # an API key with an environment name.
     # Cache misses lookup the environment name using OpsRampAPI config value.

build/vm/configure.go

@@ -21,8 +21,9 @@ func main() {
 	key := flag.String("K", "", "OpsRamp Key")
 	secret := flag.String("S", "", "OpsRamp Secret")
 	tenant := flag.String("T", "", "OpsRamp TenantID")
-	tracesAPI := flag.String("B", "", "API to Sent Traces (Defaults to Authorization API specified using -A flag if not set)")
+	tracesAPI := flag.String("B", "", "API To Sent Traces (Defaults to Authorization API specified using -A flag if not set)")
 	metricsAPI := flag.String("M", "", "API To Send Metrics (Defaults to Authorization API specified using -A flag if not set)")
+	logsAPI := flag.String("L", "", "API To Send Logs (Defaults to Authorization API specified using -A flag if not set)")
 	flag.Parse()
 
 	if *api == "" {
@@ -51,6 +52,9 @@ func main() {
 	fileContent = strings.ReplaceAll(fileContent, "<KEY>", *key)
 	fileContent = strings.ReplaceAll(fileContent, "<SECRET>", *secret)
 	fileContent = strings.ReplaceAll(fileContent, "<TENANT_ID>", *tenant)
+	if *logsAPI != "" {
+		fileContent = strings.ReplaceAll(fileContent, "<OPSRAMP_LOGS_API>", *logsAPI)
+	}
 
 	if err := os.WriteFile("/opt/opsramp/tracing-proxy/conf/config_complete.yaml", []byte(fileContent), 0o600); err != nil {
 		log.Fatal(err)

build/vm/package_directories/opt/opsramp/tracing-proxy/conf/config_complete.yaml

@@ -94,6 +94,12 @@ AddAdditionalMetadata: { "app": "default" }
 # Duration is difference between end and start time of a span
 Threshold: 500
 
+# Url to push events as logs
+LogsEndpoint: "<OPSRAMP_LOGS_API>"
+
+# If true trace proxy will send span events as logs
+SendEvents: true
+
 # EnvironmentCacheTTL is the amount of time a cache entry will live that associates
 # an API key with an environment name.
 # Cache misses lookup the environment name using OpsRampAPI config value.

cmd/tracing-proxy/main.go

@@ -150,6 +150,8 @@ func main() {
 		logrusLogger.Fatal(err)
 	}
 	retryConfig := c.GetRetryConfig()
+	logsEndpoint := c.GetLogsEndpoint()
+	sendEvents := c.GetSendEvents()
 
 	userAgentAddition := "tracing-proxy/" + CollectorVersion
 	upstreamClient, err := libtrace.NewClient(libtrace.ClientConfig{ // nolint:all
@@ -180,7 +182,9 @@ func main() {
 				MaxInterval:         retryConfig.MaxInterval,
 				MaxElapsedTime:      retryConfig.MaxElapsedTime,
 			},
-			Logger: logrusLogger,
+			Logger:       logrusLogger,
+			LogsEndpoint: logsEndpoint,
+			SendEvents:   sendEvents,
 		},
 	})
 	if err != nil {
@@ -214,7 +218,9 @@ func main() {
 				MaxInterval:         retryConfig.MaxInterval,
 				MaxElapsedTime:      retryConfig.MaxElapsedTime,
 			},
-			Logger: logrusLogger,
+			Logger:       logrusLogger,
+			LogsEndpoint: logsEndpoint,
+			SendEvents:   sendEvents,
 		},
 	})
 	if err != nil {

collect/cache/cuckoo_test.go

@@ -0,0 +1,156 @@
+package cache
+
+import (
+	"fmt"
+	"github.com/dgryski/go-wyhash"
+	"github.com/opsramp/tracing-proxy/metrics"
+	"github.com/sourcegraph/conc/pool"
+	"math/rand"
+
+	"testing"
+	"time"
+)
+
+// genID returns a random hex string of length numChars
+func genID(numChars int) string {
+	seed := 3565269841805
+
+	const charset = "abcdef0123456789"
+
+	id := make([]byte, numChars)
+	for i := 0; i < numChars; i++ {
+		id[i] = charset[int(wyhash.Rng(seed))%len(charset)]
+	}
+	return string(id)
+}
+
+// Benchmark the Add function
+func BenchmarkCuckooTraceChecker_Add(b *testing.B) {
+	traceIDs := make([]string, b.N)
+	for i := 0; i < b.N; i++ {
+		traceIDs[i] = genID(32)
+	}
+
+	c := NewCuckooTraceChecker(1000000, &metrics.NullMetrics{})
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		c.Add(traceIDs[i])
+	}
+}
+
+func BenchmarkCuckooTraceChecker_AddParallel(b *testing.B) {
+	traceIDs := make([]string, b.N)
+	for i := 0; i < b.N; i++ {
+		traceIDs[i] = genID(32)
+	}
+	const numGoroutines = 70
+
+	p := pool.New().WithMaxGoroutines(numGoroutines + 1)
+	stop := make(chan struct{})
+	p.Go(func() {
+		select {
+		case <-stop:
+			return
+		default:
+			rand.Intn(100)
+		}
+	})
+
+	c := NewCuckooTraceChecker(1000000, &metrics.NullMetrics{})
+	ch := make(chan int, numGoroutines)
+	for i := 0; i < numGoroutines; i++ {
+		p.Go(func() {
+			for n := range ch {
+				if i%10000 == 0 {
+					c.Maintain()
+				}
+				c.Add(traceIDs[n])
+			}
+		})
+	}
+	b.ResetTimer()
+	for j := 0; j < b.N; j++ {
+		ch <- j
+		if j%1000 == 0 {
+			// just give things a moment to run
+			time.Sleep(1 * time.Microsecond)
+		}
+	}
+	close(ch)
+	close(stop)
+	p.Wait()
+}
+
+func BenchmarkCuckooTraceChecker_Check(b *testing.B) {
+	fmt.Println(b.N)
+	traceIDs := make([]string, b.N)
+	for i := 0; i < b.N; i++ {
+		traceIDs[i] = genID(32)
+	}
+
+	c := NewCuckooTraceChecker(1000000, &metrics.NullMetrics{})
+	// add every other one to the filter
+	for i := 0; i < b.N; i += 2 {
+		if i%10000 == 0 {
+			c.Maintain()
+		}
+		c.Add(traceIDs[i])
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		c.Check(traceIDs[i])
+	}
+}
+
+func BenchmarkCuckooTraceChecker_CheckParallel(b *testing.B) {
+	fmt.Println(b.N)
+	traceIDs := make([]string, b.N)
+	for i := 0; i < b.N; i++ {
+		traceIDs[i] = genID(32)
+	}
+
+	c := NewCuckooTraceChecker(1000000, &metrics.NullMetrics{})
+	for i := 0; i < b.N; i += 2 {
+		if i%10000 == 0 {
+			c.Maintain()
+		}
+		c.Add(traceIDs[i])
+	}
+
+	const numGoroutines = 70
+
+	p := pool.New().WithMaxGoroutines(numGoroutines + 1)
+	stop := make(chan struct{})
+	p.Go(func() {
+		n := 0
+		select {
+		case <-stop:
+			return
+		default:
+			if n&1 == 0 {
+				c.Add(traceIDs[n])
+			}
+			n++
+			if n >= b.N {
+				n = 0
+			}
+		}
+	})
+
+	ch := make(chan int, numGoroutines)
+	for i := 0; i < numGoroutines; i++ {
+		p.Go(func() {
+			for n := range ch {
+				c.Check(traceIDs[n])
+			}
+		})
+	}
+	b.ResetTimer()
+	for j := 0; j < b.N; j++ {
+		ch <- j
+	}
+	close(ch)
+	close(stop)
+	p.Wait()
+}

collect/collect.go

@@ -40,8 +40,6 @@ func GetCollectorImplementation(c config.Config) Collector {
 	Threshold = c.GetThreshold()
 	if Threshold <= 0 {
 		Threshold = 125
-	} else {
-		Threshold = Threshold / 4
 	}
 
 	collectorType, err := c.GetCollectorType()
@@ -104,11 +102,7 @@ func (i *InMemCollector) Start() error {
 	i.cache = cache.NewInMemCache(imcConfig.CacheCapacity, i.Metrics, i.Logger)
 
 	// threshold bucket ranges
-	thresholdBuckets := []float64{Threshold}
-	for i := 2; i <= 16; i++ {
-		thresholdBuckets = append(thresholdBuckets, Threshold*float64(i))
-	}
-	thresholdBuckets = append(thresholdBuckets, 32*Threshold, 64*Threshold)
+	thresholdBuckets := []float64{Threshold / 4, Threshold / 3, Threshold / 2, Threshold, 4 * (Threshold / 3), 2 * Threshold, 4 * Threshold, 16 * Threshold, 32 * Threshold}
 
 	// listen for config reloads
 	i.Config.RegisterReloadCallback(i.sendReloadSignal)

config/config.go

@@ -98,6 +98,12 @@ type Config interface {
 	// GetThreshold is used to caliculate the apdex score
 	GetThreshold() float64
 
+	// GetLogsEnpoint is used to get endpoint for logs
+	GetLogsEndpoint() string
+
+	// Send Events as Logs
+	GetSendEvents() bool
+
 	// GetOtherConfig attempts to fill the passed in struct with the contents of
 	// a subsection of the config.   This is used by optional configurations to
 	// allow different implementations of necessary interfaces configure

config/file_config.go

@@ -65,6 +65,8 @@ type configContents struct {
 	AddHostMetadataToTrace    bool
 	AddAdditionalMetadata     map[string]string
 	Threshold                 float64
+	LogsEndpoint              string
+	SendEvents                bool
 	AddRuleReasonToTrace      bool
 	EnvironmentCacheTTL       time.Duration
 	DatasetPrefix             string
@@ -478,6 +480,20 @@ func (f *fileConfig) GetThreshold() float64 {
 	return f.conf.Threshold
 }
 
+func (f *fileConfig) GetLogsEndpoint() string {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return f.conf.LogsEndpoint
+}
+
+func (f *fileConfig) GetSendEvents() bool {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return f.conf.SendEvents
+}
+
 func (f *fileConfig) GetGRPCListenAddr() (string, error) {
 	f.mux.RLock()
 	defer f.mux.RUnlock()

config/mock.go

@@ -117,6 +117,13 @@ func (m *MockConfig) GetThreshold() float64 {
 	return 0.0
 }
 
+func (m *MockConfig) GetLogsEndpoint() string {
+	return ""
+}
+
+func (m *MockConfig) GetSendEvents() bool {
+	return false
+}
 func (m *MockConfig) GetLogrusConfig() (*LogrusLoggerConfig, error) {
 	m.Mux.RLock()
 	defer m.Mux.RUnlock()

config_complete.yaml

@@ -94,6 +94,12 @@ AddAdditionalMetadata: { "app": "default" }
 # Duration is difference between end and start time of a span
 Threshold: 500
 
+# Url to push events as logs
+LogsEndpoint: "<OPSRAMP_LOGS_API>"
+
+# If true trace proxy will send span events as logs
+SendEvents: true
+
 
 # EnvironmentCacheTTL is the amount of time a cache entry will live that associates
 # an API key with an environment name.

deploy/Dockerfile

@@ -27,7 +27,7 @@ RUN set -eux; \
     echo "${ESUM}  /usr/local/bin/tini" | sha256sum -c -; \
     chmod +x /usr/local/bin/tini
 
-ARG BUILD_ID="17.0.1"
+ARG BUILD_ID="17.0.2"
 WORKDIR /app
 
 ADD ../go.mod ../go.sum ./