序
本文主要研究一下flink的MetricQueryServiceGateway
MetricQueryServiceGateway
flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/retriever/MetricQueryServiceGateway.java
代码语言：javascript复制public interface MetricQueryServiceGateway {
CompletableFuture<MetricDumpSerialization.MetricSerializationResult> queryMetrics(Time timeout);
String getAddress();
}- MetricQueryServiceGateway定义了两个方法，一个是queryMetrics，一个是getAddress；它有一个实现类为AkkaQueryServiceGateway
AkkaQueryServiceGateway
flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/retriever/impl/AkkaQueryServiceGateway.java
代码语言：javascript复制public class AkkaQueryServiceGateway implements MetricQueryServiceGateway {
private final ActorRef queryServiceActorRef;
public AkkaQueryServiceGateway(ActorRef queryServiceActorRef) {
this.queryServiceActorRef = Preconditions.checkNotNull(queryServiceActorRef);
}
@Override
public CompletableFuture<MetricDumpSerialization.MetricSerializationResult> queryMetrics(Time timeout) {
return FutureUtils.toJava(
Patterns.ask(queryServiceActorRef, MetricQueryService.getCreateDump(), timeout.toMilliseconds())
.mapTo(ClassTag$.MODULE$.apply(MetricDumpSerialization.MetricSerializationResult.class))
);
}
@Override
public String getAddress() {
return queryServiceActorRef.path().toString();
}
}- AkkaQueryServiceGateway实现了MetricQueryServiceGateway接口，它的构造器要求传入queryServiceActorRef；queryMetrics方法ask的消息类型为MetricQueryService.CreateDump；getAddress方法返回的是queryServiceActorRef.path()
MetricQueryService
flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/dump/MetricQueryService.java
代码语言：javascript复制public class MetricQueryService extends UntypedActor {
private static final Logger LOG = LoggerFactory.getLogger(MetricQueryService.class);
public static final String METRIC_QUERY_SERVICE_NAME = "MetricQueryService";
private static final String SIZE_EXCEEDED_LOG_TEMPLATE = "{} will not be reported as the metric dump would exceed the maximum size of {} bytes.";
private static final CharacterFilter FILTER = new CharacterFilter() {
@Override
public String filterCharacters(String input) {
return replaceInvalidChars(input);
}
};
private final MetricDumpSerializer serializer = new MetricDumpSerializer();
private final Map<Gauge<?>, Tuple2<QueryScopeInfo, String>> gauges = new HashMap<>();
private final Map<Counter, Tuple2<QueryScopeInfo, String>> counters = new HashMap<>();
private final Map<Histogram, Tuple2<QueryScopeInfo, String>> histograms = new HashMap<>();
private final Map<Meter, Tuple2<QueryScopeInfo, String>> meters = new HashMap<>();
private final long messageSizeLimit;
//......
@Override
public void onReceive(Object message) {
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);
dump = enforceSizeLimit(dump);
getSender().tell(dump, getSelf());
} else {
LOG.warn("MetricQueryServiceActor received an invalid message. " message.toString());
getSender().tell(new Status.Failure(new IOException("MetricQueryServiceActor received an invalid message. " message.toString())), getSelf());
}
} catch (Exception e) {
LOG.warn("An exception occurred while processing a message.", e);
}
}
public static Object getCreateDump() {
return CreateDump.INSTANCE;
}
private static class CreateDump implements Serializable {
private static final CreateDump INSTANCE = new CreateDump();
}
//......
}- MetricQueryService继承了UntypedActor，它的onReceive方法判断message类型，如果为CreateDump的话，则调用MetricDumpSerialization.MetricDumpSerializer.serialize(counters, gauges, histograms, meters)方法来序列化metrics得到MetricDumpSerialization.MetricSerializationResult，然后使用getSender().tell(dump, getSelf())返回数据
MetricDumpSerialization
flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/dump/MetricDumpSerialization.java
代码语言：javascript复制public class MetricDumpSerialization {
//......
public static class MetricSerializationResult implements Serializable {
private static final long serialVersionUID = 6928770855951536906L;
public final byte[] serializedCounters;
public final byte[] serializedGauges;
public final byte[] serializedMeters;
public final byte[] serializedHistograms;
public final int numCounters;
public final int numGauges;
public final int numMeters;
public final int numHistograms;
public MetricSerializationResult(
byte[] serializedCounters,
byte[] serializedGauges,
byte[] serializedMeters,
byte[] serializedHistograms,
int numCounters,
int numGauges,
int numMeters,
int numHistograms) {
Preconditions.checkNotNull(serializedCounters);
Preconditions.checkNotNull(serializedGauges);
Preconditions.checkNotNull(serializedMeters);
Preconditions.checkNotNull(serializedHistograms);
Preconditions.checkArgument(numCounters >= 0);
Preconditions.checkArgument(numGauges >= 0);
Preconditions.checkArgument(numMeters >= 0);
Preconditions.checkArgument(numHistograms >= 0);
this.serializedCounters = serializedCounters;
this.serializedGauges = serializedGauges;
this.serializedMeters = serializedMeters;
this.serializedHistograms = serializedHistograms;
this.numCounters = numCounters;
this.numGauges = numGauges;
this.numMeters = numMeters;
this.numHistograms = numHistograms;
}
}
public static class MetricDumpSerializer {
private DataOutputSerializer countersBuffer = new DataOutputSerializer(1024 * 8);
private DataOutputSerializer gaugesBuffer = new DataOutputSerializer(1024 * 8);
private DataOutputSerializer metersBuffer = new DataOutputSerializer(1024 * 8);
private DataOutputSerializer histogramsBuffer = new DataOutputSerializer(1024 * 8);
public MetricSerializationResult serialize(
Map<Counter, Tuple2<QueryScopeInfo, String>> counters,
Map<Gauge<?>, Tuple2<QueryScopeInfo, String>> gauges,
Map<Histogram, Tuple2<QueryScopeInfo, String>> histograms,
Map<Meter, Tuple2<QueryScopeInfo, String>> meters) {
countersBuffer.clear();
int numCounters = 0;
for (Map.Entry<Counter, Tuple2<QueryScopeInfo, String>> entry : counters.entrySet()) {
try {
serializeCounter(countersBuffer, entry.getValue().f0, entry.getValue().f1, entry.getKey());
numCounters ;
} catch (Exception e) {
LOG.debug("Failed to serialize counter.", e);
}
}
gaugesBuffer.clear();
int numGauges = 0;
for (Map.Entry<Gauge<?>, Tuple2<QueryScopeInfo, String>> entry : gauges.entrySet()) {
try {
serializeGauge(gaugesBuffer, entry.getValue().f0, entry.getValue().f1, entry.getKey());
numGauges ;
} catch (Exception e) {
LOG.debug("Failed to serialize gauge.", e);
}
}
histogramsBuffer.clear();
int numHistograms = 0;
for (Map.Entry<Histogram, Tuple2<QueryScopeInfo, String>> entry : histograms.entrySet()) {
try {
serializeHistogram(histogramsBuffer, entry.getValue().f0, entry.getValue().f1, entry.getKey());
numHistograms ;
} catch (Exception e) {
LOG.debug("Failed to serialize histogram.", e);
}
}
metersBuffer.clear();
int numMeters = 0;
for (Map.Entry<Meter, Tuple2<QueryScopeInfo, String>> entry : meters.entrySet()) {
try {
serializeMeter(metersBuffer, entry.getValue().f0, entry.getValue().f1, entry.getKey());
numMeters ;
} catch (Exception e) {
LOG.debug("Failed to serialize meter.", e);
}
}
return new MetricSerializationResult(
countersBuffer.getCopyOfBuffer(),
gaugesBuffer.getCopyOfBuffer(),
metersBuffer.getCopyOfBuffer(),
histogramsBuffer.getCopyOfBuffer(),
numCounters,
numGauges,
numMeters,
numHistograms);
}
public void close() {
countersBuffer = null;
gaugesBuffer = null;
metersBuffer = null;
histogramsBuffer = null;
}
}
//......
}- MetricDumpSerialization有几个静态类分别是MetricSerializationResult、MetricDumpSerializer、MetricDumpDeserializer；MetricDumpSerializer提供了serialize方法用于将counters、gauges、histograms、meters指标序列化为MetricSerializationResult
小结
- MetricQueryServiceGateway定义了两个方法，一个是queryMetrics，一个是getAddress；它有一个实现类为AkkaQueryServiceGateway
- AkkaQueryServiceGateway实现了MetricQueryServiceGateway接口，它的构造器要求传入queryServiceActorRef；queryMetrics方法ask的消息类型为MetricQueryService.CreateDump；getAddress方法返回的是queryServiceActorRef.path()
- MetricQueryService继承了UntypedActor，它的onReceive方法判断message类型，如果为CreateDump的话，则调用MetricDumpSerialization.MetricDumpSerializer.serialize(counters, gauges, histograms, meters)方法来序列化metrics得到MetricDumpSerialization.MetricSerializationResult，然后使用getSender().tell(dump, getSelf())返回数据；MetricDumpSerialization有几个静态类分别是MetricSerializationResult、MetricDumpSerializer、MetricDumpDeserializer；MetricDumpSerializer提供了serialize方法用于将counters、gauges、histograms、meters指标序列化为MetricSerializationResult
doc
- MetricQueryServiceGateway
