Skip to content

ref(kafka): Remove dead Kafka message types and unused profiling writer #666

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mjq wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ref(kafka): Remove dead Kafka message types and unused profiling writer #666

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 0 additions & 22 deletions cmd/vroom/chunk.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import (
"google.golang.org/api/googleapi"

"github.com/getsentry/vroom/internal/chunk"
"github.com/getsentry/vroom/internal/platform"
"github.com/getsentry/vroom/internal/storageutil"
)

Expand Down Expand Up @@ -320,24 +319,3 @@ func (env *environment) getRawChunk(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write(b)
}

type (
ChunkKafkaMessage struct {
ProjectID uint64 `json:"project_id"`
ProfilerID string `json:"profiler_id"`
ChunkID string `json:"chunk_id"`

StartTimestamp float64 `json:"start_timestamp"`
EndTimestamp float64 `json:"end_timestamp"`
DurationMS uint64 `json:"duration_ms"`

Received float64 `json:"received"`
RetentionDays int `json:"retention_days"`

Environment string `json:"environment"`
Platform platform.Platform `json:"platform"`
Release string `json:"release"`
SDKName string `json:"sdk_name"`
SDKVersion string `json:"sdk_version"`
}
)
7 changes: 1 addition & 6 deletions cmd/vroom/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,8 @@ type (
KafkaSslKeyPath string `env:"SENTRY_KAFKA_SSL_KEY_PATH"`

OccurrencesKafkaBrokers []string `env:"SENTRY_KAFKA_BROKERS_OCCURRENCES" env-default:"localhost:9092"`
ProfilingKafkaBrokers []string `env:"SENTRY_KAFKA_BROKERS_PROFILING" env-default:"localhost:9092"`
SpansKafkaBrokers []string `env:"SENTRY_KAFKA_BROKERS_SPANS" env-default:"localhost:9092"`

CallTreesKafkaTopic string `env:"SENTRY_KAFKA_TOPIC_CALL_TREES" env-default:"profiles-call-tree"`
OccurrencesKafkaTopic string `env:"SENTRY_KAFKA_TOPIC_OCCURRENCES" env-default:"ingest-occurrences"`
ProfileChunksKafkaTopic string `env:"SENTRY_KAFKA_TOPIC_PROFILE_CHUNKS" env-default:"snuba-profile-chunks"`
ProfilesKafkaTopic string `env:"SENTRY_KAKFA_TOPIC_PROFILES" env-default:"processed-profiles"`
OccurrencesKafkaTopic string `env:"SENTRY_KAFKA_TOPIC_OCCURRENCES" env-default:"ingest-occurrences"`

BucketURL string `env:"SENTRY_BUCKET_PROFILES" env-default:"file://./test/gcs/sentry-profiles"`
}
Expand Down
49 changes: 0 additions & 49 deletions cmd/vroom/kafka.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,58 +3,9 @@ package main
import (
"context"

"github.com/getsentry/vroom/internal/nodetree"
"github.com/getsentry/vroom/internal/platform"
"github.com/segmentio/kafka-go"
)

type (
// FunctionsKafkaMessage is representing the struct we send to Kafka to insert functions in ClickHouse.
FunctionsKafkaMessage struct {
Environment string `json:"environment,omitempty"`
Functions []nodetree.CallTreeFunction `json:"functions"`
ID string `json:"profile_id"`
Platform platform.Platform `json:"platform"`
ProjectID uint64 `json:"project_id"`
Received int64 `json:"received"`
Release string `json:"release,omitempty"`
RetentionDays int `json:"retention_days"`
Timestamp int64 `json:"timestamp"`
TransactionName string `json:"transaction_name"`
StartTimestamp float64 `json:"start_timestamp,omitempty"`
EndTimestamp float64 `json:"end_timestamp,omitempty"`
ProfilingType string `json:"profiling_type,omitempty"`
MaterializationVersion uint8 `json:"materialization_version"`
}

// ProfileKafkaMessage is representing the struct we send to Kafka to insert a profile in ClickHouse.
ProfileKafkaMessage struct {
AndroidAPILevel uint32 `json:"android_api_level,omitempty"`
Architecture string `json:"architecture,omitempty"`
DeviceClassification string `json:"device_classification,omitempty"`
DeviceLocale string `json:"device_locale"`
DeviceManufacturer string `json:"device_manufacturer"`
DeviceModel string `json:"device_model"`
DeviceOSBuildNumber string `json:"device_os_build_number,omitempty"`
DeviceOSName string `json:"device_os_name"`
DeviceOSVersion string `json:"device_os_version"`
DurationNS uint64 `json:"duration_ns"`
Environment string `json:"environment,omitempty"`
ID string `json:"profile_id"`
OrganizationID uint64 `json:"organization_id"`
Platform platform.Platform `json:"platform"`
ProjectID uint64 `json:"project_id"`
Received int64 `json:"received"`
RetentionDays int `json:"retention_days"`
SDKName string `json:"sdk_name,omitempty"`
SDKVersion string `json:"sdk_version,omitempty"`
TraceID string `json:"trace_id"`
TransactionID string `json:"transaction_id"`
TransactionName string `json:"transaction_name"`
VersionCode string `json:"version_code"`
VersionName string `json:"version_name"`
}
)
type KafkaWriter interface {
WriteMessages(ctx context.Context, msgs ...kafka.Message) error
Close() error
Expand Down
21 changes: 0 additions & 21 deletions cmd/vroom/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@ type environment struct {
config ServiceConfig

occurrencesWriter KafkaWriter
profilingWriter KafkaWriter

storage *blob.Bucket
}
Expand All @@ -43,11 +42,6 @@ var (
readJobs chan storageutil.ReadJob
)

const (
KiB int64 = 1024
MiB = 1024 * KiB
)

func newEnvironment() (*environment, error) {
var e environment
err := cleanenv.ReadEnv(&e.config)
Expand All @@ -71,17 +65,6 @@ func newEnvironment() (*environment, error) {
WriteTimeout: 3 * time.Second,
Transport: createKafkaRoundTripper(e.config),
}
e.profilingWriter = &kafka.Writer{
Addr: kafka.TCP(e.config.ProfilingKafkaBrokers...),
Async: true,
Balancer: kafka.CRC32Balancer{},
BatchBytes: 20 * MiB,
BatchSize: 10,
Compression: kafka.Lz4,
ReadTimeout: 3 * time.Second,
WriteTimeout: 3 * time.Second,
Transport: createKafkaRoundTripper(e.config),
}
return &e, nil
}

Expand All @@ -94,10 +77,6 @@ func (e *environment) shutdown() {
if err != nil {
sentry.CaptureException(err)
}
err = e.profilingWriter.Close()
if err != nil {
sentry.CaptureException(err)
}
sentry.Flush(5 * time.Second)
}

Expand Down
1 change: 0 additions & 1 deletion devservices/config.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ services:
ports:
- 127.0.0.1:8085:8085
environment:
SENTRY_KAFKA_BROKERS_PROFILING: kafka-kafka-1:9092
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
Expand Down
Loading