ref(regressed): Remove Kafka occurrence production

cmd/vroom/chunk_test.go

@@ -6,8 +6,6 @@ import (
 	"os"
 	"testing"
 
-	"github.com/segmentio/kafka-go"
-
 	"gocloud.dev/blob"
 	_ "gocloud.dev/blob/fileblob"
 )
@@ -38,13 +36,3 @@ func TestMain(m *testing.M) {
 
 	os.Exit(code)
 }
-
-type KafkaWriterMock struct{}
-
-func (k KafkaWriterMock) WriteMessages(_ context.Context, _ ...kafka.Message) error {
-	return nil
-}
-
-func (k KafkaWriterMock) Close() error {
-	return nil
-}

cmd/vroom/config.go

@@ -8,17 +8,6 @@ type (
 
 		SentryDSN string `env:"SENTRY_DSN"`
 
-		KafkaSaslMechanism string `env:"SENTRY_KAFKA_SASL_MECHANISM"`
-		KafkaSaslUsername  string `env:"SENTRY_KAFKA_SASL_USERNAME"`
-		KafkaSaslPassword  string `env:"SENTRY_KAFKA_SASL_PASSWORD"`
-		KafkaSslCaPath     string `env:"SENTRY_KAFKA_SSL_CA_PATH"`
-		KafkaSslCertPath   string `env:"SENTRY_KAFKA_SSL_CERT_PATH"`
-		KafkaSslKeyPath    string `env:"SENTRY_KAFKA_SSL_KEY_PATH"`
-
-		OccurrencesKafkaBrokers []string `env:"SENTRY_KAFKA_BROKERS_OCCURRENCES" env-default:"localhost:9092"`
-
-		OccurrencesKafkaTopic string `env:"SENTRY_KAFKA_TOPIC_OCCURRENCES" env-default:"ingest-occurrences"`
-
 		BucketURL string `env:"SENTRY_BUCKET_PROFILES" env-default:"file://./test/gcs/sentry-profiles"`
 	}
 )

cmd/vroom/main.go

@@ -16,7 +16,6 @@ import (
 	sentryhttp "github.com/getsentry/sentry-go/http"
 	"github.com/ilyakaznacheev/cleanenv"
 	"github.com/julienschmidt/httprouter"
-	"github.com/segmentio/kafka-go"
 	"gocloud.dev/blob"
 	_ "gocloud.dev/blob/azureblob"
 	_ "gocloud.dev/blob/fileblob"
@@ -32,8 +31,6 @@ import (
 type environment struct {
 	config ServiceConfig
 
-	occurrencesWriter KafkaWriter
-
 	storage *blob.Bucket
 }
 
@@ -54,16 +51,6 @@ func newEnvironment(ctx context.Context) (*environment, error) {
 		return nil, err
 	}
 
-	e.occurrencesWriter = &kafka.Writer{
-		Addr:         kafka.TCP(e.config.OccurrencesKafkaBrokers...),
-		Async:        true,
-		Balancer:     kafka.CRC32Balancer{},
-		BatchSize:    100,
-		ReadTimeout:  3 * time.Second,
-		Topic:        e.config.OccurrencesKafkaTopic,
-		WriteTimeout: 3 * time.Second,
-		Transport:    createKafkaRoundTripper(e.config),
-	}
 	return &e, nil
 }
 
@@ -72,10 +59,6 @@ func (e *environment) shutdown(ctx context.Context) {
 	if err != nil {
 		sentry.CaptureException(ctx, err)
 	}
-	err = e.occurrencesWriter.Close()
-	if err != nil {
-		sentry.CaptureException(ctx, err)
-	}
 	sentry.Flush(5 * time.Second)
 }
 

cmd/vroom/regressed.go

@@ -11,7 +11,6 @@ import (
 
 func (env *environment) postRegressed(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
-	produceToKafka := r.URL.Query().Get("produce") != "false"
 
 	regressedFunctions, err := decodeRegressedFunctionPayload(ctx, r)
 	if err != nil {
@@ -41,11 +40,8 @@ func (env *environment) postRegressed(w http.ResponseWriter, r *http.Request) {
 	data := struct {
 		Occurrences        int                            `json:"occurrences"`
 		Emitted            []occurrence.RegressedFunction `json:"emitted"`
-		OccurrencePayloads *[]*occurrence.Occurrence      `json:"occurrence_payloads,omitempty"`
-	}{Occurrences: len(occurrences), Emitted: emitted}
-	if !produceToKafka {
-		data.OccurrencePayloads = &occurrences
-	}
+		OccurrencePayloads []*occurrence.Occurrence       `json:"occurrence_payloads"`
+	}{Occurrences: len(occurrences), Emitted: emitted, OccurrencePayloads: occurrences}
 	b, err := json.Marshal(data)
 	s.Finish()
 	if err != nil {
@@ -54,25 +50,6 @@ func (env *environment) postRegressed(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if produceToKafka {
-		occurrenceMessages, err := occurrence.GenerateKafkaMessageBatch(occurrences)
-		if err != nil {
-			sentry.CaptureException(ctx, err)
-			w.WriteHeader(http.StatusInternalServerError)
-			return
-		}
-
-		s = sentry.StartSpan(ctx, "processing")
-		s.Description = "Send occurrences to Kafka"
-		err = env.occurrencesWriter.WriteMessages(ctx, occurrenceMessages...)
-		s.Finish()
-		if err != nil {
-			sentry.CaptureException(ctx, err)
-			w.WriteHeader(http.StatusInternalServerError)
-			return
-		}
-	}
-
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(b)

cmd/vroom/regressed_test.go

@@ -1,9 +1,7 @@
 package main
 
 import (
-	"context"
 	"encoding/json"
-	"errors"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -15,25 +13,8 @@ import (
 	"github.com/getsentry/vroom/internal/profile"
 	"github.com/getsentry/vroom/internal/sample"
 	"github.com/getsentry/vroom/internal/storageutil"
-	"github.com/segmentio/kafka-go"
 )
 
-type regressedKafkaWriter struct {
-	calls    int
-	messages []kafka.Message
-	err      error
-}
-
-func (w *regressedKafkaWriter) WriteMessages(_ context.Context, messages ...kafka.Message) error {
-	w.calls++
-	w.messages = append(w.messages, messages...)
-	return w.err
-}
-
-func (w *regressedKafkaWriter) Close() error {
-	return nil
-}
-
 func TestPostRegressed(t *testing.T) {
 	p := profile.New(&sample.Profile{
 		Platform: platform.Node,
@@ -59,35 +40,24 @@ func TestPostRegressed(t *testing.T) {
 		name       string
 		query      string
 		body       string
-		writerErr  error
-		wantCalls  int
 		wantCount  int
-		wantInline bool
 		wantStatus int
 	}{
-		{name: "default", body: payload, wantCalls: 1, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "empty", query: "?produce=", body: payload, wantCalls: 1, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "bare", query: "?produce", body: payload, wantCalls: 1, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "true", query: "?produce=true", body: payload, wantCalls: 1, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "zero", query: "?produce=0", body: payload, wantCalls: 1, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "uppercase", query: "?produce=False", body: payload, wantCalls: 1, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "false", query: "?produce=false", body: payload, wantInline: true, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "false empty batch", query: "?produce=false", body: "[]", wantInline: true, wantStatus: http.StatusOK},
-		{name: "writer failure", body: payload, writerErr: errors.New("write failed"), wantCalls: 1, wantStatus: http.StatusInternalServerError},
-		{name: "false ignores writer failure", query: "?produce=false", body: payload, writerErr: errors.New("write failed"), wantInline: true, wantCount: 1, wantStatus: http.StatusOK},
-		{name: "invalid payload", query: "?produce=false", body: "invalid", wantStatus: http.StatusBadRequest},
+		{name: "default", body: payload, wantCount: 1, wantStatus: http.StatusOK},
+		{name: "legacy false ignored", query: "?produce=false", body: payload, wantCount: 1, wantStatus: http.StatusOK},
+		{name: "legacy true ignored", query: "?produce=true", body: payload, wantCount: 1, wantStatus: http.StatusOK},
+		{name: "empty batch", body: "[]", wantStatus: http.StatusOK},
+		{name: "null batch", body: "null", wantStatus: http.StatusOK},
+		{name: "missing frame", body: strings.Replace(payload, "1176630088", "0", 1), wantStatus: http.StatusOK},
+		{name: "invalid payload", body: "invalid", wantStatus: http.StatusBadRequest},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
-			writer := &regressedKafkaWriter{err: tt.writerErr}
-			env := environment{occurrencesWriter: writer}
+			env := environment{}
 			rec := httptest.NewRecorder()
 			env.postRegressed(rec, httptest.NewRequest(http.MethodPost, "/regressed"+tt.query, strings.NewReader(tt.body)))
 			if rec.Code != tt.wantStatus {
 				t.Fatalf("status = %d, want %d", rec.Code, tt.wantStatus)
 			}
-			if writer.calls != tt.wantCalls {
-				t.Fatalf("Kafka calls = %d, want %d", writer.calls, tt.wantCalls)
-			}
 			if tt.wantStatus != http.StatusOK {
 				return
 			}
@@ -105,25 +75,13 @@ func TestPostRegressed(t *testing.T) {
 			if response.Occurrences != tt.wantCount || len(response.Emitted) != tt.wantCount {
 				t.Fatalf("occurrences = %d, emitted = %d, want %d", response.Occurrences, len(response.Emitted), tt.wantCount)
 			}
-			if (response.OccurrencePayloads != nil) != tt.wantInline {
-				t.Fatalf("occurrence_payloads presence mismatch: %s", rec.Body.String())
-			}
-			var occurrences []*occurrence.Occurrence
-			if tt.wantInline {
-				occurrences = response.OccurrencePayloads
-			} else {
-				for _, message := range writer.messages {
-					var o occurrence.Occurrence
-					if err := json.Unmarshal(message.Value, &o); err != nil {
-						t.Fatal(err)
-					}
-					occurrences = append(occurrences, &o)
-				}
+			if response.OccurrencePayloads == nil || response.Emitted == nil {
+				t.Fatalf("expected response arrays, got: %s", rec.Body.String())
 			}
-			if len(occurrences) != tt.wantCount {
-				t.Fatalf("payload count = %d, want %d", len(occurrences), tt.wantCount)
+			if len(response.OccurrencePayloads) != tt.wantCount {
+				t.Fatalf("payload count = %d, want %d", len(response.OccurrencePayloads), tt.wantCount)
 			}
-			for _, o := range occurrences {
+			for _, o := range response.OccurrencePayloads {
 				if o.ProjectID != 2 || o.Culprit != "work" || o.ID == "" || o.PayloadType != occurrence.OccurrencePayload {
 					t.Fatalf("unexpected occurrence: %+v", o)
 				}

devservices/config.yml

@@ -2,23 +2,16 @@ x-sentry-service-config:
   version: 0.1
   service_name: vroom
   dependencies:
-    kafka:
-      description: Shared instance of kafka used by sentry services
-      remote:
-        repo_name: sentry-shared-kafka
-        branch: main
-        repo_link: https://github.com/getsentry/sentry-shared-kafka.git
     vroom:
       description: Sentry's profiling service, processing and deriving data about your profiles
   modes:
-    default: [kafka, vroom]
+    default: [vroom]
 services:
   vroom:
     image: ghcr.io/getsentry/vroom:nightly
     ports:
       - 127.0.0.1:8085:8085
     environment:
-      SENTRY_KAFKA_BROKERS_OCCURRENCES: kafka-kafka-1:9092
       SENTRY_BUCKET_PROFILES: file://localhost//var/lib/sentry-profiles
       SENTRY_SNUBA_HOST: http://127.0.0.1:1218
     volumes:

go.mod

@@ -18,7 +18,6 @@ require (
 	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5
 	github.com/pierrec/lz4 v2.6.1+incompatible
 	github.com/pierrec/lz4/v4 v4.1.15
-	github.com/segmentio/kafka-go v0.4.38
 	gocloud.dev v0.46.0
 	google.golang.org/api v0.272.0
 )
@@ -88,8 +87,6 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/sirupsen/logrus v1.9.1 // indirect
 	github.com/spiffe/go-spiffe/v2 v2.7.0 // indirect
-	github.com/xdg/scram v1.0.5 // indirect
-	github.com/xdg/stringprep v1.0.3 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect