diff options
author | Austin S. Hemmelgarn <austin@netdata.cloud> | 2024-02-13 06:56:20 -0500 |
---|---|---|
committer | GitHub <noreply@github.com> | 2024-02-13 06:56:20 -0500 |
commit | 3a29b66132f561c910d827e8c7ae82997f7c1f30 (patch) | |
tree | a9306156631b6b188de8877f7c1dbdbe8b067804 /src/go/collectors/go.d.plugin/modules/pulsar | |
parent | 57eec3da0e51baa400037ccc4b547cb839ab6ffa (diff) |
Include Go plugin sources in main repository. (#16997)
* Include Go plugin sources in main repository.
* Fix CI issues.
* Rename source tree.
Diffstat (limited to 'src/go/collectors/go.d.plugin/modules/pulsar')
13 files changed, 4989 insertions, 0 deletions
diff --git a/src/go/collectors/go.d.plugin/modules/pulsar/README.md b/src/go/collectors/go.d.plugin/modules/pulsar/README.md new file mode 120000 index 0000000000..dfa55301c8 --- /dev/null +++ b/src/go/collectors/go.d.plugin/modules/pulsar/README.md @@ -0,0 +1 @@ +integrations/apache_pulsar.md
\ No newline at end of file diff --git a/src/go/collectors/go.d.plugin/modules/pulsar/charts.go b/src/go/collectors/go.d.plugin/modules/pulsar/charts.go new file mode 100644 index 0000000000..6dbb7b49a5 --- /dev/null +++ b/src/go/collectors/go.d.plugin/modules/pulsar/charts.go @@ -0,0 +1,663 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +package pulsar + +import ( + "fmt" + "strings" + + "github.com/netdata/go.d.plugin/pkg/prometheus" + + "github.com/netdata/go.d.plugin/agent/module" +) + +type ( + Charts = module.Charts + Chart = module.Chart + Dims = module.Dims + Dim = module.Dim + Opts = module.Opts +) + +var summaryCharts = Charts{ + sumBrokerComponentsChart.Copy(), + + sumMessagesRateChart.Copy(), + sumThroughputRateChart.Copy(), + + sumStorageSizeChart.Copy(), + sumStorageOperationsRateChart.Copy(), // optional + sumMsgBacklogSizeChart.Copy(), + sumStorageWriteLatencyChart.Copy(), + sumEntrySizeChart.Copy(), + + sumSubsDelayedChart.Copy(), + sumSubsMsgRateRedeliverChart.Copy(), // optional + sumSubsBlockedOnUnackedMsgChart.Copy(), // optional + + sumReplicationRateChart.Copy(), // optional + sumReplicationThroughputRateChart.Copy(), // optional + sumReplicationBacklogChart.Copy(), // optional +} + +var ( + sumBrokerComponentsChart = Chart{ + ID: "broker_components", + Title: "Broker Components", + Units: "components", + Fam: "ns summary", + Ctx: "pulsar.broker_components", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: "pulsar_namespaces_count", Name: "namespaces"}, + {ID: metricPulsarTopicsCount, Name: "topics"}, + {ID: metricPulsarSubscriptionsCount, Name: "subscriptions"}, + {ID: metricPulsarProducersCount, Name: "producers"}, + {ID: metricPulsarConsumersCount, Name: "consumers"}, + }, + } + sumMessagesRateChart = Chart{ + ID: "messages_rate", + Title: "Messages Rate", + Units: "messages/s", + Fam: "ns summary", + Ctx: "pulsar.messages_rate", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarRateIn, Name: "publish", Div: 1000}, + {ID: metricPulsarRateOut, Name: "dispatch", Mul: -1, Div: 1000}, + }, + } + sumThroughputRateChart = Chart{ + ID: "throughput_rate", + Title: "Throughput Rate", + Units: "KiB/s", + Fam: "ns summary", + Ctx: "pulsar.throughput_rate", + Type: module.Area, + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarThroughputIn, Name: "publish", Div: 1024 * 1000}, + {ID: metricPulsarThroughputOut, Name: "dispatch", Mul: -1, Div: 1024 * 1000}, + }, + } + sumStorageSizeChart = Chart{ + ID: "storage_size", + Title: "Storage Size", + Units: "KiB", + Fam: "ns summary", + Ctx: "pulsar.storage_size", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarStorageSize, Name: "used", Div: 1024}, + }, + } + sumStorageOperationsRateChart = Chart{ + ID: "storage_operations_rate", + Title: "Storage Read/Write Operations Rate", + Units: "message batches/s", + Fam: "ns summary", + Ctx: "pulsar.storage_operations_rate", + Type: module.Area, + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarStorageReadRate, Name: "read", Div: 1000}, + {ID: metricPulsarStorageWriteRate, Name: "write", Mul: -1, Div: 1000}, + }, + } + sumMsgBacklogSizeChart = Chart{ + ID: "msg_backlog", + Title: "Messages Backlog Size", + Units: "messages", + Fam: "ns summary", + Ctx: "pulsar.msg_backlog", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarMsgBacklog, Name: "backlog"}, + }, + } + sumStorageWriteLatencyChart = Chart{ + ID: "storage_write_latency", + Title: "Storage Write Latency", + Units: "entries/s", + Fam: "ns summary", + Ctx: "pulsar.storage_write_latency", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: "pulsar_storage_write_latency_le_0_5", Name: "<=0.5ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_1", Name: "<=1ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_5", Name: "<=5ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_10", Name: "<=10ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_20", Name: "<=20ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_50", Name: "<=50ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_100", Name: "<=100ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_200", Name: "<=200ms", Div: 60}, + {ID: "pulsar_storage_write_latency_le_1000", Name: "<=1s", Div: 60}, + {ID: "pulsar_storage_write_latency_overflow", Name: ">1s", Div: 60}, + }, + } + sumEntrySizeChart = Chart{ + ID: "entry_size", + Title: "Entry Size", + Units: "entries/s", + Fam: "ns summary", + Ctx: "pulsar.entry_size", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: "pulsar_entry_size_le_128", Name: "<=128B", Div: 60}, + {ID: "pulsar_entry_size_le_512", Name: "<=512B", Div: 60}, + {ID: "pulsar_entry_size_le_1_kb", Name: "<=1KB", Div: 60}, + {ID: "pulsar_entry_size_le_2_kb", Name: "<=2KB", Div: 60}, + {ID: "pulsar_entry_size_le_4_kb", Name: "<=4KB", Div: 60}, + {ID: "pulsar_entry_size_le_16_kb", Name: "<=16KB", Div: 60}, + {ID: "pulsar_entry_size_le_100_kb", Name: "<=100KB", Div: 60}, + {ID: "pulsar_entry_size_le_1_mb", Name: "<=1MB", Div: 60}, + {ID: "pulsar_entry_size_le_overflow", Name: ">1MB", Div: 60}, + }, + } + sumSubsDelayedChart = Chart{ + ID: "subscription_delayed", + Title: "Subscriptions Delayed for Dispatching", + Units: "message batches", + Fam: "ns summary", + Ctx: "pulsar.subscription_delayed", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarSubscriptionDelayed, Name: "delayed"}, + }, + } + sumSubsMsgRateRedeliverChart = Chart{ + ID: "subscription_msg_rate_redeliver", + Title: "Subscriptions Redelivered Message Rate", + Units: "messages/s", + Fam: "ns summary", + Ctx: "pulsar.subscription_msg_rate_redeliver", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarSubscriptionMsgRateRedeliver, Name: "redelivered", Div: 1000}, + }, + } + sumSubsBlockedOnUnackedMsgChart = Chart{ + ID: "subscription_blocked_on_unacked_messages", + Title: "Subscriptions Blocked On Unacked Messages", + Units: "subscriptions", + Fam: "ns summary", + Ctx: "pulsar.subscription_blocked_on_unacked_messages", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarSubscriptionBlockedOnUnackedMessages, Name: "blocked"}, + }, + } + sumReplicationRateChart = Chart{ + ID: "replication_rate", + Title: "Replication Rate", + Units: "messages/s", + Fam: "ns summary", + Ctx: "pulsar.replication_rate", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarReplicationRateIn, Name: "in", Div: 1000}, + {ID: metricPulsarReplicationRateOut, Name: "out", Mul: -1, Div: 1000}, + }, + } + sumReplicationThroughputRateChart = Chart{ + ID: "replication_throughput_rate", + Title: "Replication Throughput Rate", + Units: "KiB/s", + Fam: "ns summary", + Ctx: "pulsar.replication_throughput_rate", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarReplicationThroughputIn, Name: "in", Div: 1024 * 1000}, + {ID: metricPulsarReplicationThroughputOut, Name: "out", Mul: -1, Div: 1024 * 1000}, + }, + } + sumReplicationBacklogChart = Chart{ + ID: "replication_backlog", + Title: "Replication Backlog", + Units: "messages", + Fam: "ns summary", + Ctx: "pulsar.replication_backlog", + Opts: Opts{StoreFirst: true}, + Dims: Dims{ + {ID: metricPulsarReplicationBacklog, Name: "backlog"}, + }, + } +) + +var namespaceCharts = Charts{ + nsBrokerComponentsChart.Copy(), + topicProducersChart.Copy(), + topicSubscriptionsChart.Copy(), + topicConsumersChart.Copy(), + + nsMessagesRateChart.Copy(), + topicMessagesRateInChart.Copy(), + topicMessagesRateOutChart.Copy(), + nsThroughputRateCharts.Copy(), + topicThroughputRateInChart.Copy(), + topicThroughputRateOutChart.Copy(), + + nsStorageSizeChart.Copy(), + topicStorageSizeChart.Copy(), + nsStorageOperationsChart.Copy(), // optional + topicStorageReadRateChart.Copy(), // optional + topicStorageWriteRateChart.Copy(), // optional + nsMsgBacklogSizeChart.Copy(), + topicMsgBacklogSizeChart.Copy(), + nsStorageWriteLatencyChart.Copy(), + nsEntrySizeChart.Copy(), + + nsSubsDelayedChart.Copy(), + topicSubsDelayedChart.Copy(), + nsSubsMsgRateRedeliverChart.Copy(), // optional + topicSubsMsgRateRedeliverChart.Copy(), // optional + nsSubsBlockedOnUnackedMsgChart.Copy(), // optional + topicSubsBlockedOnUnackedMsgChart.Copy(), // optional + + nsReplicationRateChart.Copy(), // optional + topicReplicationRateInChart.Copy(), // optional + topicReplicationRateOutChart.Copy(), // optional + nsReplicationThroughputChart.Copy(), // optional + topicReplicationThroughputRateInChart.Copy(), // optional + topicReplicationThroughputRateOutChart.Copy(), // optional + nsReplicationBacklogChart.Copy(), // optional + topicReplicationBacklogChart.Copy(), // optional +} + +func toNamespaceChart(chart Chart) Chart { + if chart.ID == sumBrokerComponentsChart.ID { + _ = chart.RemoveDim("pulsar_namespaces_count") + } + chart.ID += "_namespace_%s" + chart.Fam = "ns %s" + if idx := strings.IndexByte(chart.Ctx, '.'); idx > 0 { + // pulsar.messages_rate => pulsar.namespace_messages_rate + chart.Ctx = chart.Ctx[:idx+1] + "namespace_" + chart.Ctx[idx+1:] + } + for _, dim := range chart.Dims { + dim.ID += "_%s" + } + return chart +} + +var ( + nsBrokerComponentsChart = toNamespaceChart(sumBrokerComponentsChart) + nsMessagesRateChart = toNamespaceChart(sumMessagesRateChart) + nsThroughputRateCharts = toNamespaceChart(sumThroughputRateChart) + nsStorageSizeChart = toNamespaceChart(sumStorageSizeChart) + nsStorageOperationsChart = toNamespaceChart(sumStorageOperationsRateChart) + nsMsgBacklogSizeChart = toNamespaceChart(sumMsgBacklogSizeChart) + nsStorageWriteLatencyChart = toNamespaceChart(sumStorageWriteLatencyChart) + nsEntrySizeChart = toNamespaceChart(sumEntrySizeChart) + nsSubsDelayedChart = toNamespaceChart(sumSubsDelayedChart) + nsSubsMsgRateRedeliverChart = toNamespaceChart(sumSubsMsgRateRedeliverChart) + nsSubsBlockedOnUnackedMsgChart = toNamespaceChart(sumSubsBlockedOnUnackedMsgChart) + nsReplicationRateChart = toNamespaceChart(sumReplicationRateChart) + nsReplicationThroughputChart = toNamespaceChart(sumReplicationThroughputRateChart) + nsReplicationBacklogChart = toNamespaceChart(sumReplicationBacklogChart) + + topicProducersChart = Chart{ + ID: "topic_producers_namespace_%s", + Title: "Topic Producers", + Units: "producers", + Fam: "ns %s", + Ctx: "pulsar.topic_producers", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicSubscriptionsChart = Chart{ + ID: "topic_subscriptions_namespace_%s", + Title: "Topic Subscriptions", + Units: "subscriptions", + Fam: "ns %s", + Ctx: "pulsar.topic_subscriptions", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicConsumersChart = Chart{ + ID: "topic_consumers_namespace_%s", + Title: "Topic Consumers", + Units: "consumers", + Fam: "ns %s", + Ctx: "pulsar.topic_consumers", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicMessagesRateInChart = Chart{ + ID: "topic_messages_rate_in_namespace_%s", + Title: "Topic Publish Messages Rate", + Units: "publishes/s", + Fam: "ns %s", + Ctx: "pulsar.topic_messages_rate_in", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicMessagesRateOutChart = Chart{ + ID: "topic_messages_rate_out_namespace_%s", + Title: "Topic Dispatch Messages Rate", + Units: "dispatches/s", + Fam: "ns %s", + Ctx: "pulsar.topic_messages_rate_out", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicThroughputRateInChart = Chart{ + ID: "topic_throughput_rate_in_namespace_%s", + Title: "Topic Publish Throughput Rate", + Units: "KiB/s", + Fam: "ns %s", + Ctx: "pulsar.topic_throughput_rate_in", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicThroughputRateOutChart = Chart{ + ID: "topic_throughput_rate_out_namespace_%s", + Title: "Topic Dispatch Throughput Rate", + Units: "KiB/s", + Fam: "ns %s", + Ctx: "pulsar.topic_throughput_rate_out", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicStorageSizeChart = Chart{ + ID: "topic_storage_size_namespace_%s", + Title: "Topic Storage Size", + Units: "KiB", + Fam: "ns %s", + Ctx: "pulsar.topic_storage_size", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicStorageReadRateChart = Chart{ + ID: "topic_storage_read_rate_namespace_%s", + Title: "Topic Storage Read Rate", + Units: "message batches/s", + Fam: "ns %s", + Ctx: "pulsar.topic_storage_read_rate", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicStorageWriteRateChart = Chart{ + ID: "topic_storage_write_rate_namespace_%s", + Title: "Topic Storage Write Rate", + Units: "message batches/s", + Fam: "ns %s", + Ctx: "pulsar.topic_storage_write_rate", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicMsgBacklogSizeChart = Chart{ + ID: "topic_msg_backlog_namespace_%s", + Title: "Topic Messages Backlog Size", + Units: "messages", + Fam: "ns %s", + Ctx: "pulsar.topic_msg_backlog", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicSubsDelayedChart = Chart{ + ID: "topic_subscription_delayed_namespace_%s", + Title: "Topic Subscriptions Delayed for Dispatching", + Units: "message batches", + Fam: "ns %s", + Ctx: "pulsar.topic_subscription_delayed", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicSubsMsgRateRedeliverChart = Chart{ + ID: "topic_subscription_msg_rate_redeliver_namespace_%s", + Title: "Topic Subscriptions Redelivered Message Rate", + Units: "messages/s", + Fam: "ns %s", + Ctx: "pulsar.topic_subscription_msg_rate_redeliver", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicSubsBlockedOnUnackedMsgChart = Chart{ + ID: "topic_subscription_blocked_on_unacked_messages_namespace_%s", + Title: "Topic Subscriptions Blocked On Unacked Messages", + Units: "blocked subscriptions", + Fam: "ns %s", + Ctx: "pulsar.topic_subscription_blocked_on_unacked_messages", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicReplicationRateInChart = Chart{ + ID: "topic_replication_rate_in_namespace_%s", + Title: "Topic Replication Rate From Remote Cluster", + Units: "messages/s", + Fam: "ns %s", + Ctx: "pulsar.topic_replication_rate_in", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicReplicationRateOutChart = Chart{ + ID: "replication_rate_out_namespace_%s", + Title: "Topic Replication Rate To Remote Cluster", + Units: "messages/s", + Fam: "ns %s", + Ctx: "pulsar.topic_replication_rate_out", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicReplicationThroughputRateInChart = Chart{ + ID: "topic_replication_throughput_rate_in_namespace_%s", + Title: "Topic Replication Throughput Rate From Remote Cluster", + Units: "KiB/s", + Fam: "ns %s", + Ctx: "pulsar.topic_replication_throughput_rate_in", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicReplicationThroughputRateOutChart = Chart{ + ID: "topic_replication_throughput_rate_out_namespace_%s", + Title: "Topic Replication Throughput Rate To Remote Cluster", + Units: "KiB/s", + Fam: "ns %s", + Ctx: "pulsar.topic_replication_throughput_rate_out", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } + topicReplicationBacklogChart = Chart{ + ID: "topic_replication_backlog_namespace_%s", + Title: "Topic Replication Backlog", + Units: "messages", + Fam: "ns %s", + Ctx: "pulsar.topic_replication_backlog", + Type: module.Stacked, + Opts: Opts{StoreFirst: true}, + } +) + +func (p *Pulsar) adjustCharts(pms prometheus.Series) { + if pms := pms.FindByName(metricPulsarStorageReadRate); pms.Len() == 0 || pms[0].Labels.Get("namespace") == "" { + p.removeSummaryChart(sumStorageOperationsRateChart.ID) + p.removeNamespaceChart(nsStorageOperationsChart.ID) + p.removeNamespaceChart(topicStorageReadRateChart.ID) + p.removeNamespaceChart(topicStorageWriteRateChart.ID) + delete(p.topicChartsMapping, topicStorageReadRateChart.ID) + delete(p.topicChartsMapping, topicStorageWriteRateChart.ID) + } + if pms.FindByName(metricPulsarSubscriptionMsgRateRedeliver).Len() == 0 { + p.removeSummaryChart(sumSubsMsgRateRedeliverChart.ID) + p.removeSummaryChart(sumSubsBlockedOnUnackedMsgChart.ID) + p.removeNamespaceChart(nsSubsMsgRateRedeliverChart.ID) + p.removeNamespaceChart(nsSubsBlockedOnUnackedMsgChart.ID) + p.removeNamespaceChart(topicSubsMsgRateRedeliverChart.ID) + p.removeNamespaceChart(topicSubsBlockedOnUnackedMsgChart.ID) + delete(p.topicChartsMapping, topicSubsMsgRateRedeliverChart.ID) + delete(p.topicChartsMapping, topicSubsBlockedOnUnackedMsgChart.ID) + } + if pms.FindByName(metricPulsarReplicationBacklog).Len() == 0 { + p.removeSummaryChart(sumReplicationRateChart.ID) + p.removeSummaryChart(sumReplicationThroughputRateChart.ID) + p.removeSummaryChart(sumReplicationBacklogChart.ID) + p.removeNamespaceChart(nsReplicationRateChart.ID) + p.removeNamespaceChart(nsReplicationThroughputChart.ID) + p.removeNamespaceChart(nsReplicationBacklogChart.ID) + p.removeNamespaceChart(topicReplicationRateInChart.ID) + p.removeNamespaceChart(topicReplicationRateOutChart.ID) + p.removeNamespaceChart(topicReplicationThroughputRateInChart.ID) + p.removeNamespaceChart(topicReplicationThroughputRateOutChart.ID) + p.removeNamespaceChart(topicReplicationBacklogChart.ID) + delete(p.topicChartsMapping, topicReplicationRateInChart.ID) + delete(p.topicChartsMapping, topicReplicationRateOutChart.ID) + delete(p.topicChartsMapping, topicReplicationThroughputRateInChart.ID) + delete(p.topicChartsMapping, topicReplicationThroughputRateOutChart.ID) + delete(p.topicChartsMapping, topicReplicationBacklogChart.ID) + } +} + +func (p *Pulsar) removeSummaryChart(chartID string) { + if err := p.Charts().Remove(chartID); err != nil { + p.Warning(err) + } +} + +func (p *Pulsar) removeNamespaceChart(chartID string) { + if err := p.nsCharts.Remove(chartID); err != nil { + p.Warning(err) + } +} + +func (p *Pulsar) updateCharts() { + // NOTE: order is important + for ns := range p.curCache.namespaces { + if !p.cache.namespaces[ns] { + p.cache.namespaces[ns] = true + p.addNamespaceCharts(ns) + } + } + for top := range p.curCache.topics { + if !p.cache.topics[top] { + p.cache.topics[top] = true + p.addTopicToCharts(top) + } + } + for top := range p.cache.topics { + if p.curCache.topics[top] { + continue + } + delete(p.cache.topics, top) + p.removeTopicFromCharts(top) + } + for ns := range p.cache.namespaces { + if p.curCache.namespaces[ns] { + continue + } + delete(p.cache.namespaces, ns) + p.removeNamespaceFromCharts(ns) + } +} + +func (p *Pulsar) addNamespaceCharts(ns namespace) { + charts := p.nsCharts.Copy() + for _, chart := range *charts { + chart.ID = fmt.Sprintf(chart.ID, ns.name) + chart.Fam = fmt.Sprintf(chart.Fam, ns.name) + for _, dim := range chart.Dims { + dim.ID = fmt.Sprintf(dim.ID, ns.name) + } + } + if err := p.Charts().Add(*charts...); err != nil { + p.Warning(err) + } +} + +func (p *Pulsar) removeNamespaceFromCharts(ns namespace) { + for _, chart := range *p.nsCharts { + id := fmt.Sprintf(chart.ID, ns.name) + if chart = p.Charts().Get(id); chart != nil { + chart.MarkRemove() + } else { + p.Warningf("could not remove namespace chart '%s'", id) + } + } +} + +func (p *Pulsar) addTopicToCharts(top topic) { + for id, metric := range p.topicChartsMapping { + id = fmt.Sprintf(id, top.namespace) + chart := p.Charts().Get(id) + if chart == nil { + p.Warningf("could not add topic '%s' to chart '%s': chart not found", top.name, id) + continue + } + + dim := Dim{ID: metric + "_" + top.name, Name: extractTopicName(top)} + switch metric { + case metricPulsarThroughputIn, + metricPulsarThroughputOut, + metricPulsarReplicationThroughputIn, + metricPulsarReplicationThroughputOut: + dim.Div = 1024 * 1000 + case metricPulsarRateIn, + metricPulsarRateOut, + metricPulsarStorageWriteRate, + metricPulsarStorageReadRate, + metricPulsarSubscriptionMsgRateRedeliver, + metricPulsarReplicationRateIn, + metricPulsarReplicationRateOut: + dim.Div = 1000 + case metricPulsarStorageSize: + dim.Div = 1024 + } + + if err := chart.AddDim(&dim); err != nil { + p.Warning(err) + } + chart.MarkNotCreated() + } +} + +func (p *Pulsar) removeTopicFromCharts(top topic) { + for id, metric := range p.topicChartsMapping { + id = fmt.Sprintf(id, top.namespace) + chart := p.Charts().Get(id) + if chart == nil { + p.Warningf("could not remove topic '%s' from chart '%s': chart not found", top.name, id) + continue + } + + if err := chart.MarkDimRemove(metric+"_"+top.name, true); err != nil { + p.Warning(err) + } + chart.MarkNotCreated() + } +} + +func topicChartsMapping() map[string]string { + return map[string]string{ + topicSubscriptionsChart.ID: metricPulsarSubscriptionsCount, + topicProducersChart.ID: metricPulsarProducersCount, + topicConsumersChart.ID: metricPulsarConsumersCount, + topicMessagesRateInChart.ID: metricPulsarRateIn, + topicMessagesRateOutChart.ID: metricPulsarRateOut, + topicThroughputRateInChart.ID: metricPulsarThroughputIn, + topicThroughputRateOutChart.ID: metricPulsarThroughputOut, + topicStorageSizeChart.ID: metricPulsarStorageSize, + topicStorageReadRateChart.ID: metricPulsarStorageReadRate, + topicStorageWriteRateChart.ID: metricPulsarStorageWriteRate, + topicMsgBacklogSizeChart.ID: metricPulsarMsgBacklog, + topicSubsDelayedChart.ID: metricPulsarSubscriptionDelayed, + topicSubsMsgRateRedeliverChart.ID: metricPulsarSubscriptionMsgRateRedeliver, + topicSubsBlockedOnUnackedMsgChart.ID: metricPulsarSubscriptionBlockedOnUnackedMessages, + topicReplicationRateInChart.ID: metricPulsarReplicationRateIn, + topicReplicationRateOutChart.ID: metricPulsarReplicationRateOut, + topicReplicationThroughputRateInChart.ID: metricPulsarReplicationThroughputIn, + topicReplicationThroughputRateOutChart.ID: metricPulsarReplicationThroughputOut, + topicReplicationBacklogChart.ID: metricPulsarReplicationBacklog, + } +} + +func extractTopicName(top topic) string { + // persistent://sample/ns1/demo-1 => p:demo-1 + if idx := strings.LastIndexByte(top.name, '/'); idx > 0 { + return top.name[:1] + ":" + top.name[idx+1:] + } + return top.name +} diff --git a/src/go/collectors/go.d.plugin/modules/pulsar/collect.go b/src/go/collectors/go.d.plugin/modules/pulsar/collect.go new file mode 100644 index 0000000000..6c0b93ce00 --- /dev/null +++ b/src/go/collectors/go.d.plugin/modules/pulsar/collect.go @@ -0,0 +1,138 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +package pulsar + +import ( + "errors" + "strings" + + "github.com/netdata/go.d.plugin/pkg/prometheus" + "github.com/netdata/go.d.plugin/pkg/stm" +) + +func isValidPulsarMetrics(pms prometheus.Series) bool { + return pms.FindByName(metricPulsarTopicsCount).Len() > 0 +} + +func (p *Pulsar) resetCurCache() { + for ns := range p.curCache.namespaces { + delete(p.curCache.namespaces, ns) + } + for top := range p.curCache.topics { + delete(p.curCache.topics, top) + } +} + +func (p *Pulsar) collect() (map[string]int64, error) { + pms, err := p.prom.ScrapeSeries() + if err != nil { + return nil, err + } + + if !isValidPulsarMetrics(pms) { + return nil, errors.New("returned metrics aren't Apache Pulsar metrics") + } + + p.once.Do(func() { + p.adjustCharts(pms) + }) + + mx := p.collectMetrics(pms) + p.updateCharts() + p.resetCurCache() + + return stm.ToMap(mx), nil +} + +func (p *Pulsar) collectMetrics(pms prometheus.Series) map[string]float64 { + mx := make(map[string]float64) + p.collectBroker(mx, pms) + return mx +} + +func (p *Pulsar) collectBroker(mx map[string]float64, pms prometheus.Series) { + pms = findPulsarMetrics(pms) + for _, pm := range pms { + ns, top := newNamespace(pm), newTopic(pm) + if ns.name == "" { + continue + } + + p.curCache.namespaces[ns] = true + + value := pm.Value * precision(pm.Name()) + mx[pm.Name()] += value + mx[pm.Name()+"_"+ns.name] += value + + if top.name == "" || !p.topicFilter.MatchString(top.name) { + continue + } + + p.curCache.topics[top] = true + mx[pm.Name()+"_"+top.name] += value + } + mx["pulsar_namespaces_count"] = float64(len(p.curCache.namespaces)) +} + +func newNamespace(pm prometheus.SeriesSample) namespace { + return namespace{ + name: pm.Labels.Get("namespace"), + } +} + +func newTopic(pm prometheus.SeriesSample) topic { + return topic{ + namespace: pm.Labels.Get("namespace"), + name: pm.Labels.Get("topic"), + } +} + +func findPulsarMetrics(pms prometheus.Series) prometheus.Series { + var ms prometheus.Series + for _, pm := range pms { + if isPulsarHistogram(pm) { + ms = append(ms, pm) + } + } + pms = pms.FindByNames( + metricPulsarTopicsCount, + metricPulsarSubscriptionDelayed, + metricPulsarSubscriptionsCount, + metricPulsarProducersCount, + metricPulsarConsumersCount, + metricPulsarRateIn, + metricPulsarRateOut, + metricPulsarThroughputIn, + metricPulsarThroughputOut, + metricPulsarStorageSize, + metricPulsarStorageWriteRate, + metricPulsarStorageReadRate, + metricPulsarMsgBacklog, + metricPulsarSubscriptionMsgRateRedeliver, + metricPulsarSubscriptionBlockedOnUnackedMessages, + ) + return append(ms, pms...) +} + +func isPulsarHistogram(pm prometheus.SeriesSample) bool { + s := pm.Name() + return strings.HasPrefix(s, "pulsar_storage_write_latency") || strings.HasPrefix(s, "pulsar_entry_size") +} + +func precision(metric string) float64 { + switch metric { + case metricPulsarRateIn, + metricPulsarRateOut, + metricPulsarThroughputIn, + metricPulsarThroughputOut, + metricPulsarStorageWriteRate, + metricPulsarStorageReadRate, + metricPulsarSubscriptionMsgRateRedeliver, + metricPulsarReplicationRateIn, + metricPulsarReplicationRateOut, + metricPulsarReplicationThroughputIn, + metricPulsarReplicationThroughputOut: + return 1000 + } + return 1 +} diff --git a/src/go/collectors/go.d.plugin/modules/pulsar/config_schema.json b/src/go/collectors/go.d.plugin/modules/pulsar/config_schema.json new file mode 100644 index 0000000000..083eb0b981 --- /dev/null +++ b/src/go/collectors/go.d.plugin/modules/pulsar/config_schema.json @@ -0,0 +1,76 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "go.d/pulsar job configuration schema.", + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "url": { + "type": "string" + }, + "timeout": { + "type": [ + "string", + "integer" + ] + }, + "topic_filter": { + "type": "object", + "properties": { + "includes": { + "type": "array", + "items": { + "type": "string" + } + }, + "excludes": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, + "username": { + "type": "string" + }, + "password": { + "type": "string" + }, + "proxy_url": { + "type": "string" + }, + "proxy_username": { + "type": "string" + }, + "proxy_password": { + "type": "string" + }, + "headers": { + "type": "object", + "additionalProperties": { + "type": "string" + } + }, + "not_follow_redirects": { + "type": "boolean" + }, + "tls_ca": { + "type": "string" + }, + "tls_cert": { + "type": "string" + }, + "tls_key": { + "type": "string" + }, + "insecure_skip_verify": { + "type": "boolean" + } + }, + "required": [ + "name", + "url" + ] +} diff --git a/src/go/collectors/go.d.plugin/modules/pulsar/integrations/apache_pulsar.md b/src/go/collectors/go.d.plugin/modules/pulsar/integrations/apach |