flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-10247) Run MetricQueryService in separate thread pool
Date Mon, 10 Sep 2018 15:56:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-10247?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16609426#comment-16609426
] 

ASF GitHub Bot commented on FLINK-10247:
----------------------------------------

yanghua commented on a change in pull request #6676: [FLINK-10247][Metrics] Run MetricQueryService
in separate thread pool
URL: https://github.com/apache/flink/pull/6676#discussion_r216376070
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/metrics/dump/MetricQueryService.java
 ##########
 @@ -221,4 +193,75 @@ public static Object getCreateDump() {
 	private static class CreateDump implements Serializable {
 		private static final CreateDump INSTANCE = new CreateDump();
 	}
+
+	/**
+	 * This runnable executes add metric, remove metric and create dump logic after notified.
+	 */
+	private static final class MetricMessageHandlerRunnable implements Runnable {
+		private final Object message;
+		private final Map<Gauge<?>, Tuple2<QueryScopeInfo, String>> gauges;
+		private final Map<Counter, Tuple2<QueryScopeInfo, String>> counters;
+		private final Map<Histogram, Tuple2<QueryScopeInfo, String>> histograms;
+		private final Map<Meter, Tuple2<QueryScopeInfo, String>> meters;
+		private final ActorRef sender;
+		private final ActorRef self;
+		private final MetricDumpSerializer serializer;
+
+		public MetricMessageHandlerRunnable(Object message, Map<Gauge<?>, Tuple2<QueryScopeInfo,
String>> gauges,
+			Map<Counter, Tuple2<QueryScopeInfo, String>> counters, Map<Histogram, Tuple2<QueryScopeInfo,
String>> histograms,
+			Map<Meter, Tuple2<QueryScopeInfo, String>> meters, ActorRef sender, ActorRef
self,
+			MetricDumpSerializer serializer) {
+			this.message = message;
+			this.gauges = gauges;
+			this.counters = counters;
+			this.histograms = histograms;
+			this.meters = meters;
+			this.sender = sender;
+			this.self = self;
+			this.serializer = serializer;
+		}
+
+		@Override public void run() {
+			try {
+				if (message instanceof AddMetric) {
+					AddMetric added = (AddMetric) message;
+
+					String metricName = added.metricName;
+					Metric metric = added.metric;
+					AbstractMetricGroup group = added.group;
+
+					QueryScopeInfo info = group.getQueryServiceMetricInfo(FILTER);
+
+					if (metric instanceof Counter) {
+						counters.put((Counter) metric, new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+					} else if (metric instanceof Gauge) {
+						gauges.put((Gauge<?>) metric, new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+					} else if (metric instanceof Histogram) {
+						histograms.put((Histogram) metric, new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+					} else if (metric instanceof Meter) {
+						meters.put((Meter) metric, new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+					}
+				} else if (message instanceof RemoveMetric) {
+					Metric metric = (((RemoveMetric) message).metric);
+					if (metric instanceof Counter) {
+						this.counters.remove(metric);
+					} else if (metric instanceof Gauge) {
+						this.gauges.remove(metric);
+					} else if (metric instanceof Histogram) {
+						this.histograms.remove(metric);
+					} else if (metric instanceof Meter) {
+						this.meters.remove(metric);
+					}
+				} else if (message instanceof CreateDump) {
+					MetricDumpSerialization.MetricSerializationResult dump = serializer.serialize(counters,
gauges, histograms, meters);
+					sender.tell(dump, self);
+				} else {
+					LOG.warn("MetricQueryServiceActor received an invalid message. " + message.toString());
+					sender.tell(new Status.Failure(new IOException("MetricQueryServiceActor received an
invalid message. " + message.toString())), self);
 
 Review comment:
   the same

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


> Run MetricQueryService in separate thread pool
> ----------------------------------------------
>
>                 Key: FLINK-10247
>                 URL: https://issues.apache.org/jira/browse/FLINK-10247
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Metrics
>    Affects Versions: 1.5.3, 1.6.0, 1.7.0
>            Reporter: Till Rohrmann
>            Assignee: Shimin Yang
>            Priority: Critical
>              Labels: pull-request-available
>             Fix For: 1.6.1, 1.7.0, 1.5.4
>
>
> In order to make the {{MetricQueryService}} run independently of the main Flink components,
it should get its own dedicated thread pool assigned.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message