Skip to content
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

Jump to bottom

Add durable_name to streaming subscription metrics #81

Merged
merged 2 commits into from
Jun 6, 2019
Merged

Add durable_name to streaming subscription metrics #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b3b9fce
Add durable_name to streaming subscription metrics
Will2817 Jun 6, 2019
432bba2
Split durable queue_name
Will2817 Jun 6, 2019
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
24 changes: 21 additions & 3 deletions collector/collector_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,6 +393,7 @@ func TestStreamingSubscriptionsMetricLabels(t *testing.T) {

queueName := "some-queue-name"
durableSubscriptionName := "some-durable-name"
durableGroupSubscriptionName := "some-group-durable-name"

sc, err := stan.Connect(stanClusterName, stanClientName,
stan.NatsURL(fmt.Sprintf("nats://localhost:%d", pet.ClientPort)))
Expand All @@ -415,6 +416,13 @@ func TestStreamingSubscriptionsMetricLabels(t *testing.T) {
subscriptions = append(subscriptions, subscription)
}
subscription, err = sc.QueueSubscribe("bar", queueName, func(_ *stan.Msg) {},
stan.DurableName(durableGroupSubscriptionName))
if err != nil {
t.Fatalf("Unexpected error on subscribe: %v", err)
} else {
subscriptions = append(subscriptions, subscription)
}
subscription, err = sc.Subscribe("baz", func(_ *stan.Msg) {},
stan.DurableName(durableSubscriptionName))
if err != nil {
t.Fatalf("Unexpected error on subscribe: %v", err)
Expand Down Expand Up @@ -445,9 +453,9 @@ func TestStreamingSubscriptionsMetricLabels(t *testing.T) {
t.Fatalf("No sufficient info found for metric: %v", streamingSunscriptionMetric)
}

foundQueuedDurableLabels := false
foundQueuedDurableLabels, foundDurableLabels := false, false
expectedLabelNames := []string{"server_id", "channel", "client_id", "inbox",
"queue_name", "is_durable", "is_offline"}
"queue_name", "is_durable", "is_offline", "durable_name"}
for subscriptionIndex := range subscriptions {
expectedLabelsNotFound := make([]string, 0)
for _, labelName := range expectedLabelNames {
Expand All @@ -461,14 +469,24 @@ func TestStreamingSubscriptionsMetricLabels(t *testing.T) {
streamingSunscriptionMetric, labelMaps[subscriptionIndex]["channel"], expectedLabelsNotFound)
}

if labelMaps[subscriptionIndex]["queue_name"] == fmt.Sprintf("%v:%v", durableSubscriptionName, queueName) &&
if labelMaps[subscriptionIndex]["queue_name"] == queueName &&
labelMaps[subscriptionIndex]["durable_name"] == durableGroupSubscriptionName &&
labelMaps[subscriptionIndex]["is_durable"] == "true" {
foundQueuedDurableLabels = true
}

if labelMaps[subscriptionIndex]["durable_name"] == durableSubscriptionName &&
labelMaps[subscriptionIndex]["is_durable"] == "true" {
foundDurableLabels = true
}
}
if !foundQueuedDurableLabels {
t.Fatalf("Streaming subscription metric %v is missing expected label values "+
"for a queued durable subscription", streamingSunscriptionMetric)
}
if !foundDurableLabels {
t.Fatalf("Streaming subscription metric %v is missing expected label values "+
"for a durable subscription", streamingSunscriptionMetric)
}
}
}
13 changes: 11 additions & 2 deletions collector/streaming.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package collector
import (
"net/http"
"strconv"
"strings"
"sync"
"time"

Expand Down Expand Up @@ -149,7 +150,7 @@ type channelsCollector struct {
}

func newChannelsCollector(servers []*CollectedServer) prometheus.Collector {
subsVariableLabels := []string{"server_id", "channel", "client_id", "inbox", "queue_name", "is_durable", "is_offline"}
subsVariableLabels := []string{"server_id", "channel", "client_id", "inbox", "queue_name", "is_durable", "is_offline", "durable_name"}
nc := &channelsCollector{
httpClient: http.DefaultClient,
chanBytesTotal: prometheus.NewDesc(
Expand Down Expand Up @@ -226,8 +227,16 @@ func (nc *channelsCollector) Collect(ch chan<- prometheus.Metric) {
ch <- prometheus.MustNewConstMetric(nc.chanLastSeq, prometheus.GaugeValue, float64(channel.LastSeq), server.ID, channel.Name)

for _, sub := range channel.Subscriptions {

// If this is a durable queue group subscription then split the durable name from the queue name
durableName := sub.DurableName
queueName := sub.QueueName
if sub.IsDurable && queueName != "" {
subStrings := strings.Split(queueName, ":")
durableName, queueName = subStrings[0], subStrings[1]
}
labelValues := []string{server.ID, channel.Name, sub.ClientID, sub.Inbox,
sub.QueueName, strconv.FormatBool(sub.IsDurable), strconv.FormatBool(sub.IsOffline)}
queueName, strconv.FormatBool(sub.IsDurable), strconv.FormatBool(sub.IsOffline), durableName}
ch <- prometheus.MustNewConstMetric(nc.subsLastSent, prometheus.GaugeValue, float64(sub.LastSent), labelValues...)
ch <- prometheus.MustNewConstMetric(nc.subsPendingCount, prometheus.GaugeValue, float64(sub.PendingCount), labelValues...)
ch <- prometheus.MustNewConstMetric(nc.subsMaxInFlight, prometheus.GaugeValue, float64(sub.MaxInflight), labelValues...)
Expand Down