Skip to content

Commit ea024c4

Browse files
packethogpackethog
committed
flows: update enricher
1 parent 5c3ffab commit ea024c4

3 files changed

Lines changed: 92 additions & 37 deletions

File tree

‎telemetry/enricher/Dockerfile‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
# syntax=docker/dockerfile:1
2+
3+
# Stage 1: Build the Go binary
4+
FROM golang:1.24 AS builder
5+
6+
WORKDIR /app
7+
8+
# Copy the go.mod and go.sum files to leverage Docker cache
9+
COPY go.mod go.sum ./
10+
RUN go mod download
11+
12+
# Copy the rest of the source code
13+
COPY . .
14+
15+
# Build the enricher application
16+
# Using CGO_ENABLED=0 to build a statically linked binary
17+
RUN CGO_ENABLED=0 GOOS=linux go build -o /app/enricher ./telemetry/enricher/cmd/enricher
18+
19+
# Stage 2: Create the final, minimal image
20+
FROM debian:bullseye-slim
21+
22+
WORKDIR /app
23+
24+
# Install ca-certificates for HTTPS and other TLS-based communication
25+
RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/*
26+
27+
# Copy the built binary from the builder stage
28+
COPY --from=builder /app/enricher /app/enricher
29+
30+
# Set the entrypoint for the container
31+
ENTRYPOINT ["/app/enricher"]

‎telemetry/enricher/cmd/enricher/main.go‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,13 +36,13 @@ func main() {
3636
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
3737
defer stop()
3838

39-
if os.Getenv("CLICKHOUSE_PASS") == "" {
40-
log.Fatalf("CLICKHOUSE_PASS env var not set")
41-
}
39+
// if os.Getenv("CLICKHOUSE_PASS") == "" {
40+
// log.Fatalf("CLICKHOUSE_PASS env var not set")
41+
// }
4242

43-
if os.Getenv("REDPANDA_PASS") == "" {
44-
log.Fatalf("REDPANDA_PASS env var not set")
45-
}
43+
// if os.Getenv("REDPANDA_PASS") == "" {
44+
// log.Fatalf("REDPANDA_PASS env var not set")
45+
// }
4646

4747
opts := []enricher.EnricherOption{
4848
enricher.WithClickhouseAddr(*clickhouseAddr),
@@ -55,7 +55,9 @@ func main() {
5555
enricher.WithRedpandaMetrics(true),
5656
}
5757
enricher := enricher.NewEnricher(opts...)
58+
log.Println("starting enricher...")
5859
if err := enricher.Run(ctx); err != nil {
5960
log.Fatalf("error while running enricher: %v", err)
6061
}
62+
log.Println("enricher stopped")
6163
}

‎telemetry/enricher/internal/enricher/enricher.go‎

Lines changed: 53 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,16 @@ import (
1111
"context"
1212
"crypto/tls"
1313
"database/sql"
14-
"encoding/json"
1514
"fmt"
1615
"log"
1716
"net/http"
1817

1918
"github.com/ClickHouse/clickhouse-go/v2"
2019
"github.com/twmb/franz-go/pkg/kgo"
21-
"github.com/twmb/franz-go/pkg/sasl/scram"
20+
"github.com/twmb/franz-go/pkg/sasl/aws"
2221
"github.com/twmb/franz-go/plugin/kprom"
22+
23+
awsconfig "github.com/aws/aws-sdk-go-v2/config"
2324
)
2425

2526
type EnricherOption func(*Enricher)
@@ -160,11 +161,31 @@ func (e *Enricher) Run(ctx context.Context) error {
160161
rpOpts := []kgo.Opt{}
161162
rpOpts = append(rpOpts,
162163
kgo.SeedBrokers(e.rpBroker),
163-
kgo.SASL(scram.Auth{User: e.rpUser, Pass: e.rpPass}.AsSha256Mechanism()),
164+
//kgo.SASL(scram.Auth{User: e.rpUser, Pass: e.rpPass}.AsSha256Mechanism()),
165+
kgo.SASL(aws.ManagedStreamingIAM(func(ctx context.Context) (aws.Auth, error) {
166+
cfg, err := awsconfig.LoadDefaultConfig(ctx)
167+
if err != nil {
168+
return aws.Auth{}, fmt.Errorf("failed to load aws config: %w", err)
169+
}
170+
171+
// Retrieve the temporary credentials
172+
creds, err := cfg.Credentials.Retrieve(ctx)
173+
if err != nil {
174+
return aws.Auth{}, fmt.Errorf("failed to retrieve credentials: %w", err)
175+
}
176+
177+
// Return them in the format franz-go expects
178+
return aws.Auth{
179+
AccessKey: creds.AccessKeyID,
180+
SecretKey: creds.SecretAccessKey,
181+
SessionToken: creds.SessionToken,
182+
}, nil
183+
})),
164184
kgo.SeedBrokers(e.rpBroker),
165185
kgo.ConsumeTopics(e.rpConsumerTopic),
166186
kgo.ConsumerGroup(e.rpConsumerGroup),
167187
kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
188+
kgo.DialTLS(),
168189
)
169190
if e.rpTLS {
170191
rpOpts = append(rpOpts, kgo.DialTLSConfig(new(tls.Config)))
@@ -176,7 +197,7 @@ func (e *Enricher) Run(ctx context.Context) error {
176197

177198
go func() {
178199
http.Handle("/metrics", metrics.Handler())
179-
log.Fatal(http.ListenAndServe("localhost:8888", nil))
200+
log.Fatal(http.ListenAndServe("localhost:2112", nil))
180201
}()
181202
}
182203

@@ -209,32 +230,33 @@ func (e *Enricher) Run(ctx context.Context) error {
209230
})
210231
fetches.EachRecord(func(record *kgo.Record) {
211232
// unmarshal flow record
212-
flow := &FlowSample{}
213-
err := json.Unmarshal(record.Value, flow)
214-
if err != nil {
215-
log.Printf("error unmarshalling flow record: %v", err)
216-
return
217-
}
218-
// annotate flow record
219-
for _, a := range e.annotators {
220-
if err := a.Annotate(flow); err != nil {
221-
log.Printf("error annotating flow with %s: %v", a.String(), err)
222-
}
223-
}
224-
// write flow record back onto topic
225-
body, err := json.Marshal(flow)
226-
if err != nil {
227-
log.Printf("error marshaling enriched record to json: %v", err)
228-
return
229-
}
230-
record.Topic = e.rpProducerTopic
231-
record.Value = body
232-
client.Produce(ctx, record, func(record *kgo.Record, err error) {
233-
if err != nil {
234-
// TODO: metric
235-
fmt.Printf("error producing message to redpanda: %v \n", err)
236-
}
237-
})
233+
log.Println("received record for enrichment")
234+
// flow := &FlowSample{}
235+
// err := json.Unmarshal(record.Value, flow)
236+
// if err != nil {
237+
// log.Printf("error unmarshalling flow record: %v", err)
238+
// return
239+
// }
240+
// // annotate flow record
241+
// for _, a := range e.annotators {
242+
// if err := a.Annotate(flow); err != nil {
243+
// log.Printf("error annotating flow with %s: %v", a.String(), err)
244+
// }
245+
// }
246+
// // write flow record back onto topic
247+
// body, err := json.Marshal(flow)
248+
// if err != nil {
249+
// log.Printf("error marshaling enriched record to json: %v", err)
250+
// return
251+
// }
252+
// record.Topic = e.rpProducerTopic
253+
// record.Value = body
254+
// client.Produce(ctx, record, func(record *kgo.Record, err error) {
255+
// if err != nil {
256+
// // TODO: metric
257+
// fmt.Printf("error producing message to redpanda: %v \n", err)
258+
// }
259+
// })
238260
})
239261
if err := client.CommitUncommittedOffsets(ctx); err != nil {
240262
// TODO: metric
@@ -255,7 +277,7 @@ type Annotator interface {
255277
// Annotators must implement the Annotator interface.
256278
func (e *Enricher) RegisterAnnotators(ctx context.Context) error {
257279
e.annotators = []Annotator{
258-
NewIfNameAnnotator(),
280+
// NewIfNameAnnotator(),
259281
}
260282

261283
for _, a := range e.annotators {

0 commit comments

Comments
 (0)