| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package pulsar |
| 4 | |
| 5 | import ( |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | |
| 9 | "github.com/netdata/netdata/go/plugins/pkg/prometheus" |
| 10 | "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi" |
| 11 | ) |
| 12 | |
| 13 | type ( |
| 14 | Charts = collectorapi.Charts |
| 15 | Chart = collectorapi.Chart |
| 16 | Dims = collectorapi.Dims |
| 17 | Dim = collectorapi.Dim |
| 18 | Opts = collectorapi.Opts |
| 19 | ) |
| 20 | |
| 21 | var summaryCharts = Charts{ |
| 22 | sumBrokerComponentsChart.Copy(), |
| 23 | |
| 24 | sumMessagesRateChart.Copy(), |
| 25 | sumThroughputRateChart.Copy(), |
| 26 | |
| 27 | sumStorageSizeChart.Copy(), |
| 28 | sumStorageOperationsRateChart.Copy(), // optional |
| 29 | sumMsgBacklogSizeChart.Copy(), |
| 30 | sumStorageWriteLatencyChart.Copy(), |
| 31 | sumEntrySizeChart.Copy(), |
| 32 | |
| 33 | sumSubsDelayedChart.Copy(), |
| 34 | sumSubsMsgRateRedeliverChart.Copy(), // optional |
| 35 | sumSubsBlockedOnUnackedMsgChart.Copy(), // optional |
| 36 | |
| 37 | sumReplicationRateChart.Copy(), // optional |
| 38 | sumReplicationThroughputRateChart.Copy(), // optional |
| 39 | sumReplicationBacklogChart.Copy(), // optional |
| 40 | } |
| 41 | |
| 42 | var ( |
| 43 | sumBrokerComponentsChart = Chart{ |
| 44 | ID: "broker_components", |
| 45 | Title: "Broker Components", |
| 46 | Units: "components", |
| 47 | Fam: "ns summary", |
| 48 | Ctx: "pulsar.broker_components", |
| 49 | Type: collectorapi.Stacked, |
| 50 | Opts: Opts{StoreFirst: true}, |
| 51 | Dims: Dims{ |
| 52 | {ID: "pulsar_namespaces_count", Name: "namespaces"}, |
| 53 | {ID: metricPulsarTopicsCount, Name: "topics"}, |
| 54 | {ID: metricPulsarSubscriptionsCount, Name: "subscriptions"}, |
| 55 | {ID: metricPulsarProducersCount, Name: "producers"}, |
| 56 | {ID: metricPulsarConsumersCount, Name: "consumers"}, |
| 57 | }, |
| 58 | } |
| 59 | sumMessagesRateChart = Chart{ |
| 60 | ID: "messages_rate", |
| 61 | Title: "Messages Rate", |
| 62 | Units: "messages/s", |
| 63 | Fam: "ns summary", |
| 64 | Ctx: "pulsar.messages_rate", |
| 65 | Opts: Opts{StoreFirst: true}, |
| 66 | Dims: Dims{ |
| 67 | {ID: metricPulsarRateIn, Name: "publish", Div: 1000}, |
| 68 | {ID: metricPulsarRateOut, Name: "dispatch", Mul: -1, Div: 1000}, |
| 69 | }, |
| 70 | } |
| 71 | sumThroughputRateChart = Chart{ |
| 72 | ID: "throughput_rate", |
| 73 | Title: "Throughput Rate", |
| 74 | Units: "KiB/s", |
| 75 | Fam: "ns summary", |
| 76 | Ctx: "pulsar.throughput_rate", |
| 77 | Type: collectorapi.Area, |
| 78 | Opts: Opts{StoreFirst: true}, |
| 79 | Dims: Dims{ |
| 80 | {ID: metricPulsarThroughputIn, Name: "publish", Div: 1024 * 1000}, |
| 81 | {ID: metricPulsarThroughputOut, Name: "dispatch", Mul: -1, Div: 1024 * 1000}, |
| 82 | }, |
| 83 | } |
| 84 | sumStorageSizeChart = Chart{ |
| 85 | ID: "storage_size", |
| 86 | Title: "Storage Size", |
| 87 | Units: "KiB", |
| 88 | Fam: "ns summary", |
| 89 | Ctx: "pulsar.storage_size", |
| 90 | Opts: Opts{StoreFirst: true}, |
| 91 | Dims: Dims{ |
| 92 | {ID: metricPulsarStorageSize, Name: "used", Div: 1024}, |
| 93 | }, |
| 94 | } |
| 95 | sumStorageOperationsRateChart = Chart{ |
| 96 | ID: "storage_operations_rate", |
| 97 | Title: "Storage Read/Write Operations Rate", |
| 98 | Units: "message batches/s", |
| 99 | Fam: "ns summary", |
| 100 | Ctx: "pulsar.storage_operations_rate", |
| 101 | Type: collectorapi.Area, |
| 102 | Opts: Opts{StoreFirst: true}, |
| 103 | Dims: Dims{ |
| 104 | {ID: metricPulsarStorageReadRate, Name: "read", Div: 1000}, |
| 105 | {ID: metricPulsarStorageWriteRate, Name: "write", Mul: -1, Div: 1000}, |
| 106 | }, |
| 107 | } |
| 108 | sumMsgBacklogSizeChart = Chart{ |
| 109 | ID: "msg_backlog", |
| 110 | Title: "Messages Backlog Size", |
| 111 | Units: "messages", |
| 112 | Fam: "ns summary", |
| 113 | Ctx: "pulsar.msg_backlog", |
| 114 | Opts: Opts{StoreFirst: true}, |
| 115 | Dims: Dims{ |
| 116 | {ID: metricPulsarMsgBacklog, Name: "backlog"}, |
| 117 | }, |
| 118 | } |
| 119 | sumStorageWriteLatencyChart = Chart{ |
| 120 | ID: "storage_write_latency", |
| 121 | Title: "Storage Write Latency", |
| 122 | Units: "entries/s", |
| 123 | Fam: "ns summary", |
| 124 | Ctx: "pulsar.storage_write_latency", |
| 125 | Type: collectorapi.Stacked, |
| 126 | Opts: Opts{StoreFirst: true}, |
| 127 | Dims: Dims{ |
| 128 | {ID: "pulsar_storage_write_latency_le_0_5", Name: "<=0.5ms", Div: 60}, |
| 129 | {ID: "pulsar_storage_write_latency_le_1", Name: "<=1ms", Div: 60}, |
| 130 | {ID: "pulsar_storage_write_latency_le_5", Name: "<=5ms", Div: 60}, |
| 131 | {ID: "pulsar_storage_write_latency_le_10", Name: "<=10ms", Div: 60}, |
| 132 | {ID: "pulsar_storage_write_latency_le_20", Name: "<=20ms", Div: 60}, |
| 133 | {ID: "pulsar_storage_write_latency_le_50", Name: "<=50ms", Div: 60}, |
| 134 | {ID: "pulsar_storage_write_latency_le_100", Name: "<=100ms", Div: 60}, |
| 135 | {ID: "pulsar_storage_write_latency_le_200", Name: "<=200ms", Div: 60}, |
| 136 | {ID: "pulsar_storage_write_latency_le_1000", Name: "<=1s", Div: 60}, |
| 137 | {ID: "pulsar_storage_write_latency_overflow", Name: ">1s", Div: 60}, |
| 138 | }, |
| 139 | } |
| 140 | sumEntrySizeChart = Chart{ |
| 141 | ID: "entry_size", |
| 142 | Title: "Entry Size", |
| 143 | Units: "entries/s", |
| 144 | Fam: "ns summary", |
| 145 | Ctx: "pulsar.entry_size", |
| 146 | Type: collectorapi.Stacked, |
| 147 | Opts: Opts{StoreFirst: true}, |
| 148 | Dims: Dims{ |
| 149 | {ID: "pulsar_entry_size_le_128", Name: "<=128B", Div: 60}, |
| 150 | {ID: "pulsar_entry_size_le_512", Name: "<=512B", Div: 60}, |
| 151 | {ID: "pulsar_entry_size_le_1_kb", Name: "<=1KB", Div: 60}, |
| 152 | {ID: "pulsar_entry_size_le_2_kb", Name: "<=2KB", Div: 60}, |
| 153 | {ID: "pulsar_entry_size_le_4_kb", Name: "<=4KB", Div: 60}, |
| 154 | {ID: "pulsar_entry_size_le_16_kb", Name: "<=16KB", Div: 60}, |
| 155 | {ID: "pulsar_entry_size_le_100_kb", Name: "<=100KB", Div: 60}, |
| 156 | {ID: "pulsar_entry_size_le_1_mb", Name: "<=1MB", Div: 60}, |
| 157 | {ID: "pulsar_entry_size_le_overflow", Name: ">1MB", Div: 60}, |
| 158 | }, |
| 159 | } |
| 160 | sumSubsDelayedChart = Chart{ |
| 161 | ID: "subscription_delayed", |
| 162 | Title: "Subscriptions Delayed for Dispatching", |
| 163 | Units: "message batches", |
| 164 | Fam: "ns summary", |
| 165 | Ctx: "pulsar.subscription_delayed", |
| 166 | Opts: Opts{StoreFirst: true}, |
| 167 | Dims: Dims{ |
| 168 | {ID: metricPulsarSubscriptionDelayed, Name: "delayed"}, |
| 169 | }, |
| 170 | } |
| 171 | sumSubsMsgRateRedeliverChart = Chart{ |
| 172 | ID: "subscription_msg_rate_redeliver", |
| 173 | Title: "Subscriptions Redelivered Message Rate", |
| 174 | Units: "messages/s", |
| 175 | Fam: "ns summary", |
| 176 | Ctx: "pulsar.subscription_msg_rate_redeliver", |
| 177 | Opts: Opts{StoreFirst: true}, |
| 178 | Dims: Dims{ |
| 179 | {ID: metricPulsarSubscriptionMsgRateRedeliver, Name: "redelivered", Div: 1000}, |
| 180 | }, |
| 181 | } |
| 182 | sumSubsBlockedOnUnackedMsgChart = Chart{ |
| 183 | ID: "subscription_blocked_on_unacked_messages", |
| 184 | Title: "Subscriptions Blocked On Unacked Messages", |
| 185 | Units: "subscriptions", |
| 186 | Fam: "ns summary", |
| 187 | Ctx: "pulsar.subscription_blocked_on_unacked_messages", |
| 188 | Opts: Opts{StoreFirst: true}, |
| 189 | Dims: Dims{ |
| 190 | {ID: metricPulsarSubscriptionBlockedOnUnackedMessages, Name: "blocked"}, |
| 191 | }, |
| 192 | } |
| 193 | sumReplicationRateChart = Chart{ |
| 194 | ID: "replication_rate", |
| 195 | Title: "Replication Rate", |
| 196 | Units: "messages/s", |
| 197 | Fam: "ns summary", |
| 198 | Ctx: "pulsar.replication_rate", |
| 199 | Opts: Opts{StoreFirst: true}, |
| 200 | Dims: Dims{ |
| 201 | {ID: metricPulsarReplicationRateIn, Name: "in", Div: 1000}, |
| 202 | {ID: metricPulsarReplicationRateOut, Name: "out", Mul: -1, Div: 1000}, |
| 203 | }, |
| 204 | } |
| 205 | sumReplicationThroughputRateChart = Chart{ |
| 206 | ID: "replication_throughput_rate", |
| 207 | Title: "Replication Throughput Rate", |
| 208 | Units: "KiB/s", |
| 209 | Fam: "ns summary", |
| 210 | Ctx: "pulsar.replication_throughput_rate", |
| 211 | Opts: Opts{StoreFirst: true}, |
| 212 | Dims: Dims{ |
| 213 | {ID: metricPulsarReplicationThroughputIn, Name: "in", Div: 1024 * 1000}, |
| 214 | {ID: metricPulsarReplicationThroughputOut, Name: "out", Mul: -1, Div: 1024 * 1000}, |
| 215 | }, |
| 216 | } |
| 217 | sumReplicationBacklogChart = Chart{ |
| 218 | ID: "replication_backlog", |
| 219 | Title: "Replication Backlog", |
| 220 | Units: "messages", |
| 221 | Fam: "ns summary", |
| 222 | Ctx: "pulsar.replication_backlog", |
| 223 | Opts: Opts{StoreFirst: true}, |
| 224 | Dims: Dims{ |
| 225 | {ID: metricPulsarReplicationBacklog, Name: "backlog"}, |
| 226 | }, |
| 227 | } |
| 228 | ) |
| 229 | |
| 230 | var namespaceCharts = Charts{ |
| 231 | nsBrokerComponentsChart.Copy(), |
| 232 | topicProducersChart.Copy(), |
| 233 | topicSubscriptionsChart.Copy(), |
| 234 | topicConsumersChart.Copy(), |
| 235 | |
| 236 | nsMessagesRateChart.Copy(), |
| 237 | topicMessagesRateInChart.Copy(), |
| 238 | topicMessagesRateOutChart.Copy(), |
| 239 | nsThroughputRateCharts.Copy(), |
| 240 | topicThroughputRateInChart.Copy(), |
| 241 | topicThroughputRateOutChart.Copy(), |
| 242 | |
| 243 | nsStorageSizeChart.Copy(), |
| 244 | topicStorageSizeChart.Copy(), |
| 245 | nsStorageOperationsChart.Copy(), // optional |
| 246 | topicStorageReadRateChart.Copy(), // optional |
| 247 | topicStorageWriteRateChart.Copy(), // optional |
| 248 | nsMsgBacklogSizeChart.Copy(), |
| 249 | topicMsgBacklogSizeChart.Copy(), |
| 250 | nsStorageWriteLatencyChart.Copy(), |
| 251 | nsEntrySizeChart.Copy(), |
| 252 | |
| 253 | nsSubsDelayedChart.Copy(), |
| 254 | topicSubsDelayedChart.Copy(), |
| 255 | nsSubsMsgRateRedeliverChart.Copy(), // optional |
| 256 | topicSubsMsgRateRedeliverChart.Copy(), // optional |
| 257 | nsSubsBlockedOnUnackedMsgChart.Copy(), // optional |
| 258 | topicSubsBlockedOnUnackedMsgChart.Copy(), // optional |
| 259 | |
| 260 | nsReplicationRateChart.Copy(), // optional |
| 261 | topicReplicationRateInChart.Copy(), // optional |
| 262 | topicReplicationRateOutChart.Copy(), // optional |
| 263 | nsReplicationThroughputChart.Copy(), // optional |
| 264 | topicReplicationThroughputRateInChart.Copy(), // optional |
| 265 | topicReplicationThroughputRateOutChart.Copy(), // optional |
| 266 | nsReplicationBacklogChart.Copy(), // optional |
| 267 | topicReplicationBacklogChart.Copy(), // optional |
| 268 | } |
| 269 | |
| 270 | func toNamespaceChart(chart Chart) Chart { |
| 271 | chart = *chart.Copy() |
| 272 | if chart.ID == sumBrokerComponentsChart.ID { |
| 273 | _ = chart.RemoveDim("pulsar_namespaces_count") |
| 274 | } |
| 275 | chart.ID += "_namespace_%s" |
| 276 | chart.Fam = "ns %s" |
| 277 | if idx := strings.IndexByte(chart.Ctx, '.'); idx > 0 { |
| 278 | // pulsar.messages_rate => pulsar.namespace_messages_rate |
| 279 | chart.Ctx = chart.Ctx[:idx+1] + "namespace_" + chart.Ctx[idx+1:] |
| 280 | } |
| 281 | for _, dim := range chart.Dims { |
| 282 | dim.ID += "_%s" |
| 283 | } |
| 284 | return chart |
| 285 | } |
| 286 | |
| 287 | var ( |
| 288 | nsBrokerComponentsChart = toNamespaceChart(sumBrokerComponentsChart) |
| 289 | nsMessagesRateChart = toNamespaceChart(sumMessagesRateChart) |
| 290 | nsThroughputRateCharts = toNamespaceChart(sumThroughputRateChart) |
| 291 | nsStorageSizeChart = toNamespaceChart(sumStorageSizeChart) |
| 292 | nsStorageOperationsChart = toNamespaceChart(sumStorageOperationsRateChart) |
| 293 | nsMsgBacklogSizeChart = toNamespaceChart(sumMsgBacklogSizeChart) |
| 294 | nsStorageWriteLatencyChart = toNamespaceChart(sumStorageWriteLatencyChart) |
| 295 | nsEntrySizeChart = toNamespaceChart(sumEntrySizeChart) |
| 296 | nsSubsDelayedChart = toNamespaceChart(sumSubsDelayedChart) |
| 297 | nsSubsMsgRateRedeliverChart = toNamespaceChart(sumSubsMsgRateRedeliverChart) |
| 298 | nsSubsBlockedOnUnackedMsgChart = toNamespaceChart(sumSubsBlockedOnUnackedMsgChart) |
| 299 | nsReplicationRateChart = toNamespaceChart(sumReplicationRateChart) |
| 300 | nsReplicationThroughputChart = toNamespaceChart(sumReplicationThroughputRateChart) |
| 301 | nsReplicationBacklogChart = toNamespaceChart(sumReplicationBacklogChart) |
| 302 | |
| 303 | topicProducersChart = Chart{ |
| 304 | ID: "topic_producers_namespace_%s", |
| 305 | Title: "Topic Producers", |
| 306 | Units: "producers", |
| 307 | Fam: "ns %s", |
| 308 | Ctx: "pulsar.topic_producers", |
| 309 | Type: collectorapi.Stacked, |
| 310 | Opts: Opts{StoreFirst: true}, |
| 311 | } |
| 312 | topicSubscriptionsChart = Chart{ |
| 313 | ID: "topic_subscriptions_namespace_%s", |
| 314 | Title: "Topic Subscriptions", |
| 315 | Units: "subscriptions", |
| 316 | Fam: "ns %s", |
| 317 | Ctx: "pulsar.topic_subscriptions", |
| 318 | Type: collectorapi.Stacked, |
| 319 | Opts: Opts{StoreFirst: true}, |
| 320 | } |
| 321 | topicConsumersChart = Chart{ |
| 322 | ID: "topic_consumers_namespace_%s", |
| 323 | Title: "Topic Consumers", |
| 324 | Units: "consumers", |
| 325 | Fam: "ns %s", |
| 326 | Ctx: "pulsar.topic_consumers", |
| 327 | Type: collectorapi.Stacked, |
| 328 | Opts: Opts{StoreFirst: true}, |
| 329 | } |
| 330 | topicMessagesRateInChart = Chart{ |
| 331 | ID: "topic_messages_rate_in_namespace_%s", |
| 332 | Title: "Topic Publish Messages Rate", |
| 333 | Units: "publishes/s", |
| 334 | Fam: "ns %s", |
| 335 | Ctx: "pulsar.topic_messages_rate_in", |
| 336 | Type: collectorapi.Stacked, |
| 337 | Opts: Opts{StoreFirst: true}, |
| 338 | } |
| 339 | topicMessagesRateOutChart = Chart{ |
| 340 | ID: "topic_messages_rate_out_namespace_%s", |
| 341 | Title: "Topic Dispatch Messages Rate", |
| 342 | Units: "dispatches/s", |
| 343 | Fam: "ns %s", |
| 344 | Ctx: "pulsar.topic_messages_rate_out", |
| 345 | Type: collectorapi.Stacked, |
| 346 | Opts: Opts{StoreFirst: true}, |
| 347 | } |
| 348 | topicThroughputRateInChart = Chart{ |
| 349 | ID: "topic_throughput_rate_in_namespace_%s", |
| 350 | Title: "Topic Publish Throughput Rate", |
| 351 | Units: "KiB/s", |
| 352 | Fam: "ns %s", |
| 353 | Ctx: "pulsar.topic_throughput_rate_in", |
| 354 | Type: collectorapi.Stacked, |
| 355 | Opts: Opts{StoreFirst: true}, |
| 356 | } |
| 357 | topicThroughputRateOutChart = Chart{ |
| 358 | ID: "topic_throughput_rate_out_namespace_%s", |
| 359 | Title: "Topic Dispatch Throughput Rate", |
| 360 | Units: "KiB/s", |
| 361 | Fam: "ns %s", |
| 362 | Ctx: "pulsar.topic_throughput_rate_out", |
| 363 | Type: collectorapi.Stacked, |
| 364 | Opts: Opts{StoreFirst: true}, |
| 365 | } |
| 366 | topicStorageSizeChart = Chart{ |
| 367 | ID: "topic_storage_size_namespace_%s", |
| 368 | Title: "Topic Storage Size", |
| 369 | Units: "KiB", |
| 370 | Fam: "ns %s", |
| 371 | Ctx: "pulsar.topic_storage_size", |
| 372 | Type: collectorapi.Stacked, |
| 373 | Opts: Opts{StoreFirst: true}, |
| 374 | } |
| 375 | topicStorageReadRateChart = Chart{ |
| 376 | ID: "topic_storage_read_rate_namespace_%s", |
| 377 | Title: "Topic Storage Read Rate", |
| 378 | Units: "message batches/s", |
| 379 | Fam: "ns %s", |
| 380 | Ctx: "pulsar.topic_storage_read_rate", |
| 381 | Type: collectorapi.Stacked, |
| 382 | Opts: Opts{StoreFirst: true}, |
| 383 | } |
| 384 | topicStorageWriteRateChart = Chart{ |
| 385 | ID: "topic_storage_write_rate_namespace_%s", |
| 386 | Title: "Topic Storage Write Rate", |
| 387 | Units: "message batches/s", |
| 388 | Fam: "ns %s", |
| 389 | Ctx: "pulsar.topic_storage_write_rate", |
| 390 | Type: collectorapi.Stacked, |
| 391 | Opts: Opts{StoreFirst: true}, |
| 392 | } |
| 393 | topicMsgBacklogSizeChart = Chart{ |
| 394 | ID: "topic_msg_backlog_namespace_%s", |
| 395 | Title: "Topic Messages Backlog Size", |
| 396 | Units: "messages", |
| 397 | Fam: "ns %s", |
| 398 | Ctx: "pulsar.topic_msg_backlog", |
| 399 | Type: collectorapi.Stacked, |
| 400 | Opts: Opts{StoreFirst: true}, |
| 401 | } |
| 402 | topicSubsDelayedChart = Chart{ |
| 403 | ID: "topic_subscription_delayed_namespace_%s", |
| 404 | Title: "Topic Subscriptions Delayed for Dispatching", |
| 405 | Units: "message batches", |
| 406 | Fam: "ns %s", |
| 407 | Ctx: "pulsar.topic_subscription_delayed", |
| 408 | Type: collectorapi.Stacked, |
| 409 | Opts: Opts{StoreFirst: true}, |
| 410 | } |
| 411 | topicSubsMsgRateRedeliverChart = Chart{ |
| 412 | ID: "topic_subscription_msg_rate_redeliver_namespace_%s", |
| 413 | Title: "Topic Subscriptions Redelivered Message Rate", |
| 414 | Units: "messages/s", |
| 415 | Fam: "ns %s", |
| 416 | Ctx: "pulsar.topic_subscription_msg_rate_redeliver", |
| 417 | Type: collectorapi.Stacked, |
| 418 | Opts: Opts{StoreFirst: true}, |
| 419 | } |
| 420 | topicSubsBlockedOnUnackedMsgChart = Chart{ |
| 421 | ID: "topic_subscription_blocked_on_unacked_messages_namespace_%s", |
| 422 | Title: "Topic Subscriptions Blocked On Unacked Messages", |
| 423 | Units: "blocked subscriptions", |
| 424 | Fam: "ns %s", |
| 425 | Ctx: "pulsar.topic_subscription_blocked_on_unacked_messages", |
| 426 | Type: collectorapi.Stacked, |
| 427 | Opts: Opts{StoreFirst: true}, |
| 428 | } |
| 429 | topicReplicationRateInChart = Chart{ |
| 430 | ID: "topic_replication_rate_in_namespace_%s", |
| 431 | Title: "Topic Replication Rate From Remote Cluster", |
| 432 | Units: "messages/s", |
| 433 | Fam: "ns %s", |
| 434 | Ctx: "pulsar.topic_replication_rate_in", |
| 435 | Type: collectorapi.Stacked, |
| 436 | Opts: Opts{StoreFirst: true}, |
| 437 | } |
| 438 | topicReplicationRateOutChart = Chart{ |
| 439 | ID: "replication_rate_out_namespace_%s", |
| 440 | Title: "Topic Replication Rate To Remote Cluster", |
| 441 | Units: "messages/s", |
| 442 | Fam: "ns %s", |
| 443 | Ctx: "pulsar.topic_replication_rate_out", |
| 444 | Type: collectorapi.Stacked, |
| 445 | Opts: Opts{StoreFirst: true}, |
| 446 | } |
| 447 | topicReplicationThroughputRateInChart = Chart{ |
| 448 | ID: "topic_replication_throughput_rate_in_namespace_%s", |
| 449 | Title: "Topic Replication Throughput Rate From Remote Cluster", |
| 450 | Units: "KiB/s", |
| 451 | Fam: "ns %s", |
| 452 | Ctx: "pulsar.topic_replication_throughput_rate_in", |
| 453 | Type: collectorapi.Stacked, |
| 454 | Opts: Opts{StoreFirst: true}, |
| 455 | } |
| 456 | topicReplicationThroughputRateOutChart = Chart{ |
| 457 | ID: "topic_replication_throughput_rate_out_namespace_%s", |
| 458 | Title: "Topic Replication Throughput Rate To Remote Cluster", |
| 459 | Units: "KiB/s", |
| 460 | Fam: "ns %s", |
| 461 | Ctx: "pulsar.topic_replication_throughput_rate_out", |
| 462 | Type: collectorapi.Stacked, |
| 463 | Opts: Opts{StoreFirst: true}, |
| 464 | } |
| 465 | topicReplicationBacklogChart = Chart{ |
| 466 | ID: "topic_replication_backlog_namespace_%s", |
| 467 | Title: "Topic Replication Backlog", |
| 468 | Units: "messages", |
| 469 | Fam: "ns %s", |
| 470 | Ctx: "pulsar.topic_replication_backlog", |
| 471 | Type: collectorapi.Stacked, |
| 472 | Opts: Opts{StoreFirst: true}, |
| 473 | } |
| 474 | ) |
| 475 | |
| 476 | func (c *Collector) adjustCharts(pms prometheus.Series) { |
| 477 | if pms := pms.FindByName(metricPulsarStorageReadRate); pms.Len() == 0 || pms[0].Labels.Get("namespace") == "" { |
| 478 | c.removeSummaryChart(sumStorageOperationsRateChart.ID) |
| 479 | c.removeNamespaceChart(nsStorageOperationsChart.ID) |
| 480 | c.removeNamespaceChart(topicStorageReadRateChart.ID) |
| 481 | c.removeNamespaceChart(topicStorageWriteRateChart.ID) |
| 482 | delete(c.topicChartsMapping, topicStorageReadRateChart.ID) |
| 483 | delete(c.topicChartsMapping, topicStorageWriteRateChart.ID) |
| 484 | } |
| 485 | if pms.FindByName(metricPulsarSubscriptionMsgRateRedeliver).Len() == 0 { |
| 486 | c.removeSummaryChart(sumSubsMsgRateRedeliverChart.ID) |
| 487 | c.removeSummaryChart(sumSubsBlockedOnUnackedMsgChart.ID) |
| 488 | c.removeNamespaceChart(nsSubsMsgRateRedeliverChart.ID) |
| 489 | c.removeNamespaceChart(nsSubsBlockedOnUnackedMsgChart.ID) |
| 490 | c.removeNamespaceChart(topicSubsMsgRateRedeliverChart.ID) |
| 491 | c.removeNamespaceChart(topicSubsBlockedOnUnackedMsgChart.ID) |
| 492 | delete(c.topicChartsMapping, topicSubsMsgRateRedeliverChart.ID) |
| 493 | delete(c.topicChartsMapping, topicSubsBlockedOnUnackedMsgChart.ID) |
| 494 | } |
| 495 | if pms.FindByName(metricPulsarReplicationBacklog).Len() == 0 { |
| 496 | c.removeSummaryChart(sumReplicationRateChart.ID) |
| 497 | c.removeSummaryChart(sumReplicationThroughputRateChart.ID) |
| 498 | c.removeSummaryChart(sumReplicationBacklogChart.ID) |
| 499 | c.removeNamespaceChart(nsReplicationRateChart.ID) |
| 500 | c.removeNamespaceChart(nsReplicationThroughputChart.ID) |
| 501 | c.removeNamespaceChart(nsReplicationBacklogChart.ID) |
| 502 | c.removeNamespaceChart(topicReplicationRateInChart.ID) |
| 503 | c.removeNamespaceChart(topicReplicationRateOutChart.ID) |
| 504 | c.removeNamespaceChart(topicReplicationThroughputRateInChart.ID) |
| 505 | c.removeNamespaceChart(topicReplicationThroughputRateOutChart.ID) |
| 506 | c.removeNamespaceChart(topicReplicationBacklogChart.ID) |
| 507 | delete(c.topicChartsMapping, topicReplicationRateInChart.ID) |
| 508 | delete(c.topicChartsMapping, topicReplicationRateOutChart.ID) |
| 509 | delete(c.topicChartsMapping, topicReplicationThroughputRateInChart.ID) |
| 510 | delete(c.topicChartsMapping, topicReplicationThroughputRateOutChart.ID) |
| 511 | delete(c.topicChartsMapping, topicReplicationBacklogChart.ID) |
| 512 | } |
| 513 | } |
| 514 | |
| 515 | func (c *Collector) removeSummaryChart(chartID string) { |
| 516 | if err := c.Charts().Remove(chartID); err != nil { |
| 517 | c.Warning(err) |
| 518 | } |
| 519 | } |
| 520 | |
| 521 | func (c *Collector) removeNamespaceChart(chartID string) { |
| 522 | if err := c.nsCharts.Remove(chartID); err != nil { |
| 523 | c.Warning(err) |
| 524 | } |
| 525 | } |
| 526 | |
| 527 | func (c *Collector) updateCharts() { |
| 528 | // NOTE: order is important |
| 529 | for ns := range c.curCache.namespaces { |
| 530 | if !c.cache.namespaces[ns] { |
| 531 | c.cache.namespaces[ns] = true |
| 532 | c.addNamespaceCharts(ns) |
| 533 | } |
| 534 | } |
| 535 | for top := range c.curCache.topics { |
| 536 | if !c.cache.topics[top] { |
| 537 | c.cache.topics[top] = true |
| 538 | c.addTopicToCharts(top) |
| 539 | } |
| 540 | } |
| 541 | for top := range c.cache.topics { |
| 542 | if c.curCache.topics[top] { |
| 543 | continue |
| 544 | } |
| 545 | delete(c.cache.topics, top) |
| 546 | c.removeTopicFromCharts(top) |
| 547 | } |
| 548 | for ns := range c.cache.namespaces { |
| 549 | if c.curCache.namespaces[ns] { |
| 550 | continue |
| 551 | } |
| 552 | delete(c.cache.namespaces, ns) |
| 553 | c.removeNamespaceFromCharts(ns) |
| 554 | } |
| 555 | } |
| 556 | |
| 557 | func (c *Collector) addNamespaceCharts(ns namespace) { |
| 558 | charts := c.nsCharts.Copy() |
| 559 | for _, chart := range *charts { |
| 560 | chart.ID = fmt.Sprintf(chart.ID, ns.name) |
| 561 | chart.Fam = fmt.Sprintf(chart.Fam, ns.name) |
| 562 | for _, dim := range chart.Dims { |
| 563 | dim.ID = fmt.Sprintf(dim.ID, ns.name) |
| 564 | } |
| 565 | } |
| 566 | if err := c.Charts().Add(*charts...); err != nil { |
| 567 | c.Warning(err) |
| 568 | } |
| 569 | } |
| 570 | |
| 571 | func (c *Collector) removeNamespaceFromCharts(ns namespace) { |
| 572 | for _, chart := range *c.nsCharts { |
| 573 | id := fmt.Sprintf(chart.ID, ns.name) |
| 574 | if chart = c.Charts().Get(id); chart != nil { |
| 575 | chart.MarkRemove() |
| 576 | } else { |
| 577 | c.Warningf("could not remove namespace chart '%s'", id) |
| 578 | } |
| 579 | } |
| 580 | } |
| 581 | |
| 582 | func (c *Collector) addTopicToCharts(top topic) { |
| 583 | for id, metric := range c.topicChartsMapping { |
| 584 | id = fmt.Sprintf(id, top.namespace) |
| 585 | chart := c.Charts().Get(id) |
| 586 | if chart == nil { |
| 587 | c.Warningf("could not add topic '%s' to chart '%s': chart not found", top.name, id) |
| 588 | continue |
| 589 | } |
| 590 | |
| 591 | dim := Dim{ID: metric + "_" + top.name, Name: extractTopicName(top)} |
| 592 | switch metric { |
| 593 | case metricPulsarThroughputIn, |
| 594 | metricPulsarThroughputOut, |
| 595 | metricPulsarReplicationThroughputIn, |
| 596 | metricPulsarReplicationThroughputOut: |
| 597 | dim.Div = 1024 * 1000 |
| 598 | case metricPulsarRateIn, |
| 599 | metricPulsarRateOut, |
| 600 | metricPulsarStorageWriteRate, |
| 601 | metricPulsarStorageReadRate, |
| 602 | metricPulsarSubscriptionMsgRateRedeliver, |
| 603 | metricPulsarReplicationRateIn, |
| 604 | metricPulsarReplicationRateOut: |
| 605 | dim.Div = 1000 |
| 606 | case metricPulsarStorageSize: |
| 607 | dim.Div = 1024 |
| 608 | } |
| 609 | |
| 610 | if err := chart.AddDim(&dim); err != nil { |
| 611 | c.Warning(err) |
| 612 | } |
| 613 | chart.MarkNotCreated() |
| 614 | } |
| 615 | } |
| 616 | |
| 617 | func (c *Collector) removeTopicFromCharts(top topic) { |
| 618 | for id, metric := range c.topicChartsMapping { |
| 619 | id = fmt.Sprintf(id, top.namespace) |
| 620 | chart := c.Charts().Get(id) |
| 621 | if chart == nil { |
| 622 | c.Warningf("could not remove topic '%s' from chart '%s': chart not found", top.name, id) |
| 623 | continue |
| 624 | } |
| 625 | |
| 626 | if err := chart.MarkDimRemove(metric+"_"+top.name, true); err != nil { |
| 627 | c.Warning(err) |
| 628 | } |
| 629 | chart.MarkNotCreated() |
| 630 | } |
| 631 | } |
| 632 | |
| 633 | func topicChartsMapping() map[string]string { |
| 634 | return map[string]string{ |
| 635 | topicSubscriptionsChart.ID: metricPulsarSubscriptionsCount, |
| 636 | topicProducersChart.ID: metricPulsarProducersCount, |
| 637 | topicConsumersChart.ID: metricPulsarConsumersCount, |
| 638 | topicMessagesRateInChart.ID: metricPulsarRateIn, |
| 639 | topicMessagesRateOutChart.ID: metricPulsarRateOut, |
| 640 | topicThroughputRateInChart.ID: metricPulsarThroughputIn, |
| 641 | topicThroughputRateOutChart.ID: metricPulsarThroughputOut, |
| 642 | topicStorageSizeChart.ID: metricPulsarStorageSize, |
| 643 | topicStorageReadRateChart.ID: metricPulsarStorageReadRate, |
| 644 | topicStorageWriteRateChart.ID: metricPulsarStorageWriteRate, |
| 645 | topicMsgBacklogSizeChart.ID: metricPulsarMsgBacklog, |
| 646 | topicSubsDelayedChart.ID: metricPulsarSubscriptionDelayed, |
| 647 | topicSubsMsgRateRedeliverChart.ID: metricPulsarSubscriptionMsgRateRedeliver, |
| 648 | topicSubsBlockedOnUnackedMsgChart.ID: metricPulsarSubscriptionBlockedOnUnackedMessages, |
| 649 | topicReplicationRateInChart.ID: metricPulsarReplicationRateIn, |
| 650 | topicReplicationRateOutChart.ID: metricPulsarReplicationRateOut, |
| 651 | topicReplicationThroughputRateInChart.ID: metricPulsarReplicationThroughputIn, |
| 652 | topicReplicationThroughputRateOutChart.ID: metricPulsarReplicationThroughputOut, |
| 653 | topicReplicationBacklogChart.ID: metricPulsarReplicationBacklog, |
| 654 | } |
| 655 | } |
| 656 | |
| 657 | func extractTopicName(top topic) string { |
| 658 | // persistent://sample/ns1/demo-1 => p:demo-1 |
| 659 | if idx := strings.LastIndexByte(top.name, '/'); idx > 0 { |
| 660 | return top.name[:1] + ":" + top.name[idx+1:] |
| 661 | } |
| 662 | return top.name |
| 663 | } |