Skip to content

OpenTelemetry metric plugin and dual emitting #16665

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

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
Open

OpenTelemetry metric plugin and dual emitting #16665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
8d5724d
Prototyping of OpenTelemetry metric plugin
mqliang Aug 22, 2025
fe05700
Fix compound metrics
mqliang Aug 28, 2025
1423336
Make Yammer metrics as default and add more metric name parsing rules
mqliang Sep 2, 2025
f5add16
Get rid of metric name parsing
mqliang Sep 4, 2025
03baeef
fix testing
mqliang Sep 9, 2025
5cf3397
fix testing failure caused by 1.5.0 release preparation
mqliang Sep 9, 2025
954b32c
remove unused code
mqliang Sep 9, 2025
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
6 changes: 4 additions & 2 deletions pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.pinot.broker.broker.helix;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
Expand Down Expand Up @@ -309,12 +310,13 @@ public void start()
_brokerConf.getProperty(Broker.CONFIG_OF_ENABLE_TABLE_LEVEL_METRICS, Broker.DEFAULT_ENABLE_TABLE_LEVEL_METRICS),
_brokerConf.getProperty(Broker.CONFIG_OF_ALLOWED_TABLES_FOR_EMITTING_METRICS, Collections.emptyList()));
_brokerMetrics.initializeGlobalMeters();
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.VERSION, PinotVersion.VERSION_METRIC_NAME, 1);
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.VERSION, PinotVersion.VERSION_METRIC_NAME,
1, ImmutableMap.of());
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.ZK_JUTE_MAX_BUFFER,
Integer.getInteger(ZkSystemPropertyKeys.JUTE_MAXBUFFER, 0xfffff));
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.ADAPTIVE_SERVER_SELECTOR_TYPE,
_brokerConf.getProperty(Broker.AdaptiveServerSelector.CONFIG_OF_TYPE,
Broker.AdaptiveServerSelector.DEFAULT_TYPE), 1);
Broker.AdaptiveServerSelector.DEFAULT_TYPE), 1, ImmutableMap.of());
BrokerMetrics.register(_brokerMetrics);
// Set up request handling classes
_serverRoutingStatsManager = new ServerRoutingStatsManager(_brokerConf, _brokerMetrics);
Expand Down
9 changes: 8 additions & 1 deletion ...main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.net.URI;
import java.util.ArrayList;
Expand Down Expand Up @@ -65,6 +66,7 @@
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.BrokerQueryPhase;
import org.apache.pinot.common.metrics.BrokerTimer;
import org.apache.pinot.common.metrics.MetricAttributeConstants;
import org.apache.pinot.common.request.BrokerRequest;
import org.apache.pinot.common.request.DataSource;
import org.apache.pinot.common.request.Expression;
Expand Down Expand Up @@ -862,8 +864,13 @@ protected BrokerResponse doHandleRequest(long requestId, String query, SqlNodeAn
}

for (int pool : brokerResponse.getPools()) {
String poolTag = BrokerMetrics.getTagForPreferredPool(sqlNodeAndOptions.getOptions());
String poolId = String.valueOf(pool);
_brokerMetrics.addMeteredValue(BrokerMeter.POOL_QUERIES, 1,
BrokerMetrics.getTagForPreferredPool(sqlNodeAndOptions.getOptions()), String.valueOf(pool));
ImmutableList.of(poolTag, poolId),
ImmutableMap.of(MetricAttributeConstants.POOL_TAG, poolTag,
MetricAttributeConstants.POOL_ID, poolId)
);
}

brokerResponse.setRLSFiltersApplied(rlsFiltersApplied.get());
Expand Down
12 changes: 10 additions & 2 deletions .../main/java/org/apache/pinot/broker/routing/instanceselector/BalancedInstanceSelector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
*/
package org.apache.pinot.broker.routing.instanceselector;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.time.Clock;
import java.util.HashMap;
import java.util.List;
Expand All @@ -30,6 +32,7 @@
import org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.MetricAttributeConstants;
import org.apache.pinot.common.utils.HashUtil;

/// Instance selector to balance the number of segments served by each selected server instance.
Expand Down Expand Up @@ -96,8 +99,13 @@ Pair<Map<String, String>, Map<String, String>> select(List<String> segments, int
}

for (Map.Entry<Integer, Integer> entry : poolToSegmentCount.entrySet()) {
_brokerMetrics.addMeteredValue(BrokerMeter.POOL_SEG_QUERIES, entry.getValue(),
BrokerMetrics.getTagForPreferredPool(queryOptions), String.valueOf(entry.getKey()));
String poolTag = BrokerMetrics.getTagForPreferredPool(queryOptions);
String poolID = String.valueOf(entry.getKey());
_brokerMetrics.addMeteredValue(
BrokerMeter.POOL_SEG_QUERIES, entry.getValue(),
ImmutableList.of(poolTag, poolID),
ImmutableMap.of(MetricAttributeConstants.POOL_TAG, poolTag, MetricAttributeConstants.POOL_ID, poolID)
);
}
return Pair.of(segmentToSelectedInstanceMap, optionalSegmentToInstanceMap);
}
Expand Down
8 changes: 7 additions & 1 deletion ...n/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
*/
package org.apache.pinot.broker.routing.instanceselector;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Comparator;
Expand All @@ -33,6 +35,7 @@
import org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
import org.apache.pinot.common.metrics.MetricAttributeConstants;
import org.apache.pinot.common.utils.HashUtil;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;

Expand Down Expand Up @@ -146,8 +149,11 @@ private Pair<Map<String, String>, Map<String, String>> selectServers(List<String
replicaOffset = (replicaOffset + 1) % numReplicaGroups;
}
for (Map.Entry<Integer, Integer> entry : poolToSegmentCount.entrySet()) {
String poolTag = BrokerMetrics.getTagForPreferredPool(ctx.getQueryOptions());
String poolId = String.valueOf(entry.getKey());
_brokerMetrics.addMeteredValue(BrokerMeter.POOL_SEG_QUERIES, entry.getValue(),
BrokerMetrics.getTagForPreferredPool(ctx.getQueryOptions()), String.valueOf(entry.getKey()));
ImmutableList.of(poolTag, poolId),
ImmutableMap.of(MetricAttributeConstants.POOL_TAG, poolTag, MetricAttributeConstants.POOL_ID, poolId));
}
return Pair.of(segmentToSelectedInstanceMap, optionalSegmentToInstanceMap);
}
Expand Down
Loading
Loading