-
Notifications
You must be signed in to change notification settings - Fork 621
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
memdb: like phlaredb but not using fs (#3506)
- Loading branch information
1 parent
d657e80
commit ad91b32
Showing
26 changed files
with
2,933 additions
and
499 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| package ingester | ||
|
|
||
| import ( | ||
| "context" | ||
| objstore2 "github.com/grafana/pyroscope/pkg/objstore" | ||
| "github.com/thanos-io/objstore" | ||
| "io" | ||
| ) | ||
|
|
||
| type mockBucket struct { | ||
| upload func(ctx context.Context, name string, r io.Reader) error | ||
| } | ||
|
|
||
| func (m mockBucket) Close() error { | ||
|
|
||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) Iter(ctx context.Context, dir string, f func(string) error, options ...objstore.IterOption) error { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) GetRange(ctx context.Context, name string, off, length int64) (io.ReadCloser, error) { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) Exists(ctx context.Context, name string) (bool, error) { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) IsObjNotFoundErr(err error) bool { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) IsAccessDeniedErr(err error) bool { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) Attributes(ctx context.Context, name string) (objstore.ObjectAttributes, error) { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) Upload(ctx context.Context, name string, r io.Reader) error { | ||
| return m.upload(ctx, name, r) | ||
| } | ||
|
|
||
| func (m mockBucket) Delete(ctx context.Context, name string) error { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) Name() string { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func (m mockBucket) ReaderAt(ctx context.Context, filename string) (objstore2.ReaderAtCloser, error) { | ||
| //TODO implement me | ||
| panic("implement me") | ||
| } |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,169 @@ | ||
| package memdb | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "github.com/google/uuid" | ||
| profilev1 "github.com/grafana/pyroscope/api/gen/proto/go/google/v1" | ||
| typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1" | ||
| phlaremodel "github.com/grafana/pyroscope/pkg/model" | ||
| phlarelabels "github.com/grafana/pyroscope/pkg/phlaredb/labels" | ||
| schemav1 "github.com/grafana/pyroscope/pkg/phlaredb/schemas/v1" | ||
| "github.com/grafana/pyroscope/pkg/phlaredb/symdb" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/common/model" | ||
| "go.uber.org/atomic" | ||
| "math" | ||
| "sync" | ||
| ) | ||
|
|
||
| type FlushedHead struct { | ||
| Index []byte | ||
| Profiles []byte | ||
| Symbols []byte | ||
| Meta struct { | ||
| ProfileTypeNames []string | ||
| MinTimeNanos int64 | ||
| MaxTimeNanos int64 | ||
| NumSamples uint64 | ||
| NumProfiles uint64 | ||
| NumSeries uint64 | ||
| } | ||
| } | ||
|
|
||
| type Head struct { | ||
| symbols *symdb.PartitionWriter | ||
| metaLock sync.RWMutex | ||
| minTimeNanos int64 | ||
| maxTimeNanos int64 | ||
| totalSamples *atomic.Uint64 | ||
| profiles *profilesIndex | ||
| metrics *HeadMetrics | ||
| } | ||
|
|
||
| func NewHead(metrics *HeadMetrics) *Head { | ||
| h := &Head{ | ||
| metrics: metrics, | ||
| symbols: symdb.NewPartitionWriter(0, &symdb.Config{ | ||
| Version: symdb.FormatV3, | ||
| Stacktraces: symdb.StacktracesConfig{ | ||
| MaxNodesPerChunk: 4 << 20, | ||
| }, | ||
| }), | ||
| totalSamples: atomic.NewUint64(0), | ||
| minTimeNanos: math.MaxInt64, | ||
| maxTimeNanos: 0, | ||
| profiles: newProfileIndex(metrics), | ||
| } | ||
|
|
||
| return h | ||
| } | ||
|
|
||
| func (h *Head) Ingest(p *profilev1.Profile, id uuid.UUID, externalLabels []*typesv1.LabelPair) { | ||
| if len(p.Sample) == 0 { | ||
| return | ||
| } | ||
|
|
||
| // delta not supported | ||
| externalLabels = phlaremodel.Labels(externalLabels).Delete(phlaremodel.LabelNameDelta) | ||
|
|
||
| enforceLabelOrder := phlaremodel.Labels(externalLabels).Get(phlaremodel.LabelNameOrder) == phlaremodel.LabelOrderEnforced | ||
| externalLabels = phlaremodel.Labels(externalLabels).Delete(phlaremodel.LabelNameOrder) | ||
|
|
||
| lbls, seriesFingerprints := phlarelabels.CreateProfileLabels(enforceLabelOrder, p, externalLabels...) | ||
|
|
||
| metricName := phlaremodel.Labels(externalLabels).Get(model.MetricNameLabel) | ||
|
|
||
| var profileIngested bool | ||
| memProfiles := h.symbols.WriteProfileSymbols(p) | ||
| for idxType := range memProfiles { | ||
| profile := &memProfiles[idxType] | ||
| profile.ID = id | ||
| profile.SeriesFingerprint = seriesFingerprints[idxType] | ||
| profile.Samples = profile.Samples.Compact(false) | ||
|
|
||
| profile.TotalValue = profile.Samples.Sum() | ||
|
|
||
| if profile.Samples.Len() == 0 { | ||
| continue | ||
| } | ||
|
|
||
| h.profiles.Add(profile, lbls[idxType], metricName) | ||
|
|
||
| profileIngested = true | ||
| h.totalSamples.Add(uint64(profile.Samples.Len())) | ||
| h.metrics.sampleValuesIngested.WithLabelValues(metricName).Add(float64(profile.Samples.Len())) | ||
| h.metrics.sampleValuesReceived.WithLabelValues(metricName).Add(float64(len(p.Sample))) | ||
| } | ||
|
|
||
| if !profileIngested { | ||
| return | ||
| } | ||
|
|
||
| h.metaLock.Lock() | ||
| if p.TimeNanos < h.minTimeNanos { | ||
| h.minTimeNanos = p.TimeNanos | ||
| } | ||
| if p.TimeNanos > h.maxTimeNanos { | ||
| h.maxTimeNanos = p.TimeNanos | ||
| } | ||
| h.metaLock.Unlock() | ||
| } | ||
|
|
||
| func (h *Head) Flush(ctx context.Context) (res *FlushedHead, err error) { | ||
| t := prometheus.NewTimer(h.metrics.flushedBlockDurationSeconds) | ||
| defer t.ObserveDuration() | ||
|
|
||
| if res, err = h.flush(ctx); err != nil { | ||
| h.metrics.flushedBlocks.WithLabelValues("failed").Inc() | ||
| return nil, err | ||
| } | ||
|
|
||
| blockSize := len(res.Index) + len(res.Profiles) + len(res.Symbols) | ||
| h.metrics.flushedBlocks.WithLabelValues("success").Inc() | ||
| h.metrics.flushedBlockSamples.Observe(float64(res.Meta.NumSamples)) | ||
| h.metrics.flusehdBlockProfiles.Observe(float64(res.Meta.NumProfiles)) | ||
| h.metrics.flushedBlockSeries.Observe(float64(res.Meta.NumSeries)) | ||
| h.metrics.flushedBlockSizeBytes.Observe(float64(blockSize)) | ||
| h.metrics.flushedFileSizeBytes.WithLabelValues("tsdb").Observe(float64(len(res.Index))) | ||
| h.metrics.flushedFileSizeBytes.WithLabelValues("profiles.parquet").Observe(float64(len(res.Profiles))) | ||
| h.metrics.flushedFileSizeBytes.WithLabelValues("symbols.symdb").Observe(float64(len(res.Symbols))) | ||
| return res, nil | ||
| } | ||
|
|
||
| func (h *Head) flush(ctx context.Context) (*FlushedHead, error) { | ||
| var ( | ||
| err error | ||
| profiles []schemav1.InMemoryProfile | ||
| ) | ||
| res := new(FlushedHead) | ||
| res.Meta.MinTimeNanos = h.minTimeNanos | ||
| res.Meta.MaxTimeNanos = h.maxTimeNanos | ||
| res.Meta.NumSamples = h.totalSamples.Load() | ||
| res.Meta.NumSeries = uint64(h.profiles.totalSeries.Load()) | ||
|
|
||
| if res.Meta.NumSamples == 0 { | ||
| return res, nil | ||
| } | ||
|
|
||
| symbolsBuffer := bytes.NewBuffer(nil) | ||
| if err := symdb.WritePartition(h.symbols, symbolsBuffer); err != nil { | ||
| return nil, err | ||
| } | ||
| res.Symbols = symbolsBuffer.Bytes() | ||
|
|
||
| if res.Meta.ProfileTypeNames, err = h.profiles.profileTypeNames(); err != nil { | ||
| return nil, fmt.Errorf("failed to get profile type names: %w", err) | ||
| } | ||
|
|
||
| if res.Index, profiles, err = h.profiles.Flush(ctx); err != nil { | ||
| return nil, fmt.Errorf("failed to flush profiles: %w", err) | ||
| } | ||
| res.Meta.NumProfiles = uint64(len(profiles)) | ||
|
|
||
| if res.Profiles, err = WriteProfiles(h.metrics, profiles); err != nil { | ||
| return nil, fmt.Errorf("failed to write profiles parquet: %w", err) | ||
| } | ||
| return res, nil | ||
| } |
Oops, something went wrong.