Implement the FNV hash library for the probabilistic sampler

.chloggen/probabilistic-sampler-fnv-hash-enhancement.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: processor/probabilisticsampler
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Implement the FNV hash library for the probabilistic sampler.
+
+# One or more tracking issues related to the change
+issues: [16456]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

processor/probabilisticsamplerprocessor/fnvhasher.go

@@ -0,0 +1,37 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http:https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package probabilisticsamplerprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor"
+
+import (
+ "hash/fnv"
+)
+
+// computeHash creates a hash using the FNV-1a algorithm
+func computeHash(b []byte, seed uint32) uint32 {
+ hash := fnv.New32a()
+ // the implementation fnv.Write() does not return an error, see hash/fnv/fnv.go
+ _, _ = hash.Write(i32tob(seed))
+ _, _ = hash.Write(b)
+ return hash.Sum32()
+}
+
+// i32tob converts a seed to a byte array to be used as part of fnv.Write()
+func i32tob(val uint32) []byte {
+ r := make([]byte, 4)
+ for i := uint32(0); i < 4; i++ {
+ r[i] = byte((val >> (8 * i)) & 0xff)
+ }
+ return r
+}

processor/probabilisticsamplerprocessor/logsprocessor.go

@@ -16,6 +16,7 @@ package probabilisticsamplerprocessor // import "github.com/open-telemetry/opent
 
 import (
  "context"
+ "strconv"
 
  "go.opencensus.io/stats"
  "go.opencensus.io/tag"
@@ -89,21 +90,12 @@ func (lsp *logSamplerProcessor) processLogs(ctx context.Context, ld plog.Logs) (
  }
  }
 
- sampled := hash(lidBytes, lsp.hashSeed)&bitMaskHashBuckets < priority
- var err error
- if sampled {
- err = stats.RecordWithTags(
- ctx,
- []tag.Mutator{tag.Upsert(tagPolicyKey, tagPolicyValue), tag.Upsert(tagSampledKey, "true")},
- statCountLogsSampled.M(int64(1)),
- )
- } else {
- err = stats.RecordWithTags(
- ctx,
- []tag.Mutator{tag.Upsert(tagPolicyKey, tagPolicyValue), tag.Upsert(tagSampledKey, "false")},
- statCountLogsSampled.M(int64(1)),
- )
- }
+ sampled := computeHash(lidBytes, lsp.hashSeed)&bitMaskHashBuckets < priority
+ var err error = stats.RecordWithTags(
+ ctx,
+ []tag.Mutator{tag.Upsert(tagPolicyKey, tagPolicyValue), tag.Upsert(tagSampledKey, strconv.FormatBool(sampled))},
+ statCountLogsSampled.M(int64(1)),
+ )
  if err != nil {
  lsp.logger.Error(err.Error())
  }

processor/probabilisticsamplerprocessor/logsprocessor_test.go

@@ -98,7 +98,7 @@ func TestLogsSampling(t *testing.T) {
  SamplingPercentage: 50,
  AttributeSource: traceIDAttributeSource,
  },
- received: 52,
+ received: 45, // 52
  },
  {
  name: "sampling_source no sampling",
@@ -125,7 +125,7 @@ func TestLogsSampling(t *testing.T) {
  AttributeSource: recordAttributeSource,
  FromAttribute: "foo",
  },
- received: 79,
+ received: 23,
  },
  {
  name: "sampling_priority",

processor/probabilisticsamplerprocessor/tracesprocessor.go

@@ -105,21 +105,13 @@ func (tsp *traceSamplerProcessor) processTraces(ctx context.Context, td ptrace.T
  // Hashing here prevents bias due to such systems.
  tidBytes := s.TraceID()
  sampled := sp == mustSampleSpan ||
- hash(tidBytes[:], tsp.hashSeed)&bitMaskHashBuckets < tsp.scaledSamplingRate
+ computeHash(tidBytes[:], tsp.hashSeed)&bitMaskHashBuckets < tsp.scaledSamplingRate
 
- if sampled {
- _ = stats.RecordWithTags(
- ctx,
- []tag.Mutator{tag.Upsert(tagPolicyKey, "trace_id_hash"), tag.Upsert(tagSampledKey, "true")},
- statCountTracesSampled.M(int64(1)),
- )
- } else {
- _ = stats.RecordWithTags(
- ctx,
- []tag.Mutator{tag.Upsert(tagPolicyKey, "trace_id_hash"), tag.Upsert(tagSampledKey, "false")},
- statCountTracesSampled.M(int64(1)),
- )
- }
+ _ = stats.RecordWithTags(
+ ctx,
+ []tag.Mutator{tag.Upsert(tagPolicyKey, "trace_id_hash"), tag.Upsert(tagSampledKey, strconv.FormatBool(sampled))},
+ statCountTracesSampled.M(int64(1)),
+ )
  return !sampled
  })
  // Filter out empty ScopeMetrics