-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
enhance: add streaming client metrics
Signed-off-by: chyezh <[email protected]>
- Loading branch information
Showing
62 changed files
with
1,715 additions
and
271 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
22 changes: 22 additions & 0 deletions
22
internal/distributed/streaming/internal/consumer/handler.go
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,22 @@ | ||
| package consumer | ||
|
|
||
| import "github.com/milvus-io/milvus/pkg/streaming/util/message" | ||
|
|
||
| // nopCloseHandler is a handler that do nothing when close. | ||
| type nopCloseHandler struct { | ||
| message.Handler | ||
| HandleInterceptor func(msg message.ImmutableMessage, handle func(message.ImmutableMessage)) | ||
| } | ||
|
|
||
| // Handle is the callback for handling message. | ||
| func (nch nopCloseHandler) Handle(msg message.ImmutableMessage) { | ||
| if nch.HandleInterceptor != nil { | ||
| nch.HandleInterceptor(msg, nch.Handler.Handle) | ||
| return | ||
| } | ||
| nch.Handler.Handle(msg) | ||
| } | ||
|
|
||
| // Close is called after all messages are handled or handling is interrupted. | ||
| func (nch nopCloseHandler) Close() { | ||
| } |
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
79 changes: 79 additions & 0 deletions
79
internal/distributed/streaming/internal/consumer/metrics.go
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,79 @@ | ||
| package consumer | ||
|
|
||
| import ( | ||
| "github.com/prometheus/client_golang/prometheus" | ||
|
|
||
| "github.com/milvus-io/milvus/pkg/metrics" | ||
| "github.com/milvus-io/milvus/pkg/util/paramtable" | ||
| ) | ||
|
|
||
| // newConsumerMetrics creates a new producer metrics. | ||
| func newConsumerMetrics(pchannel string) *consumerMetrics { | ||
| constLabel := prometheus.Labels{ | ||
| metrics.NodeIDLabelName: paramtable.GetStringNodeID(), | ||
| metrics.WALChannelLabelName: pchannel, | ||
| } | ||
| m := &consumerMetrics{ | ||
| available: false, | ||
| clientTotal: metrics.StreamingServiceClientConsumerTotal.MustCurryWith(constLabel), | ||
| inflightTotal: metrics.StreamingServiceClientConsumeInflightTotal.With(constLabel), | ||
| bytes: metrics.StreamingServiceClientConsumeBytes.With(constLabel), | ||
| } | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Inc() | ||
| return m | ||
| } | ||
|
|
||
| // consumerMetrics is the metrics for producer. | ||
| type consumerMetrics struct { | ||
| available bool | ||
| clientTotal *prometheus.GaugeVec | ||
| inflightTotal prometheus.Gauge | ||
| bytes prometheus.Observer | ||
| } | ||
|
|
||
| // IntoUnavailable sets the producer metrics to unavailable. | ||
| func (m *consumerMetrics) IntoUnavailable() { | ||
| if !m.available { | ||
| return | ||
| } | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Inc() | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusAvailable).Dec() | ||
| m.available = false | ||
| } | ||
|
|
||
| // IntoAvailable sets the producer metrics to available. | ||
| func (m *consumerMetrics) IntoAvailable() { | ||
| if m.available { | ||
| return | ||
| } | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusAvailable).Inc() | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Dec() | ||
| m.available = true | ||
| } | ||
|
|
||
| // StartConsume starts a consume operation. | ||
| func (m *consumerMetrics) StartConsume(bytes int) consumerMetricsGuard { | ||
| m.inflightTotal.Inc() | ||
| return consumerMetricsGuard{ | ||
| metrics: m, | ||
| bytes: bytes, | ||
| } | ||
| } | ||
|
|
||
| func (m *consumerMetrics) Close() { | ||
| if m.available { | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusAvailable).Dec() | ||
| } else { | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Dec() | ||
| } | ||
| } | ||
|
|
||
| type consumerMetricsGuard struct { | ||
| metrics *consumerMetrics | ||
| bytes int | ||
| } | ||
|
|
||
| func (g consumerMetricsGuard) Finish() { | ||
| g.metrics.inflightTotal.Dec() | ||
| g.metrics.bytes.Observe(float64(g.bytes)) | ||
| } |
101 changes: 101 additions & 0 deletions
101
internal/distributed/streaming/internal/producer/metrics.go
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,101 @@ | ||
| package producer | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
|
|
||
| "github.com/milvus-io/milvus/internal/util/streamingutil/status" | ||
| "github.com/milvus-io/milvus/pkg/metrics" | ||
| "github.com/milvus-io/milvus/pkg/util/paramtable" | ||
| ) | ||
|
|
||
| // newProducerMetrics creates a new producer metrics. | ||
| func newProducerMetrics(pchannel string) *producerMetrics { | ||
| constLabel := prometheus.Labels{ | ||
| metrics.NodeIDLabelName: paramtable.GetStringNodeID(), | ||
| metrics.WALChannelLabelName: pchannel, | ||
| } | ||
| m := &producerMetrics{ | ||
| available: false, | ||
| clientTotal: metrics.StreamingServiceClientProducerTotal.MustCurryWith(constLabel), | ||
| inflightTotal: metrics.StreamingServiceClientProduceInflightTotal.With(constLabel), | ||
| bytes: metrics.StreamingServiceClientProduceBytes.MustCurryWith(constLabel), | ||
| durationSeconds: metrics.StreamingServiceClientProduceDurationSeconds.MustCurryWith(constLabel), | ||
| } | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Inc() | ||
| return m | ||
| } | ||
|
|
||
| // producerMetrics is the metrics for producer. | ||
| type producerMetrics struct { | ||
| available bool | ||
| clientTotal *prometheus.GaugeVec | ||
| inflightTotal prometheus.Gauge | ||
| bytes prometheus.ObserverVec | ||
| durationSeconds prometheus.ObserverVec | ||
| } | ||
|
|
||
| // IntoUnavailable sets the producer metrics to unavailable. | ||
| func (m *producerMetrics) IntoUnavailable() { | ||
| if !m.available { | ||
| return | ||
| } | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Inc() | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusAvailable).Dec() | ||
| m.available = false | ||
| } | ||
|
|
||
| // IntoAvailable sets the producer metrics to available. | ||
| func (m *producerMetrics) IntoAvailable() { | ||
| if m.available { | ||
| return | ||
| } | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusAvailable).Inc() | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Dec() | ||
| m.available = true | ||
| } | ||
|
|
||
| // StartProduce starts the produce metrics. | ||
| func (m *producerMetrics) StartProduce(bytes int) produceMetricsGuard { | ||
| m.inflightTotal.Inc() | ||
| return produceMetricsGuard{ | ||
| start: time.Now(), | ||
| bytes: bytes, | ||
| metrics: m, | ||
| } | ||
| } | ||
|
|
||
| func (m *producerMetrics) Close() { | ||
| if m.available { | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusAvailable).Dec() | ||
| } else { | ||
| m.clientTotal.WithLabelValues(metrics.StreamingServiceClientStatusUnavailable).Dec() | ||
| } | ||
| } | ||
|
|
||
| // produceMetricsGuard is the guard for produce metrics. | ||
| type produceMetricsGuard struct { | ||
| start time.Time | ||
| bytes int | ||
| metrics *producerMetrics | ||
| } | ||
|
|
||
| // Finish finishes the produce metrics. | ||
| func (g produceMetricsGuard) Finish(err error) { | ||
| status := parseError(err) | ||
| g.metrics.bytes.WithLabelValues(status).Observe(float64(g.bytes)) | ||
| g.metrics.durationSeconds.WithLabelValues(status).Observe(time.Since(g.start).Seconds()) | ||
| g.metrics.inflightTotal.Dec() | ||
| } | ||
|
|
||
| // parseError parses the error to status. | ||
| func parseError(err error) string { | ||
| if err == nil { | ||
| return metrics.StreamingServiceClientStatusOK | ||
| } | ||
| if status.IsCanceled(err) { | ||
| return metrics.StreamingServiceClientStatusCancel | ||
| } | ||
| return metrics.StreamignServiceClientStatusError | ||
| } |
Oops, something went wrong.