Hi Chesnay,
Thanks for the prompt response and feedback, it's very much
appreciated. Please see the inline responses below to your questions:
*Was there anything in the logs (ideally on debug)?*
I didn't see anything within the logs that seemed to indicate anything
out of the ordinary. I'm currently using a MiniClusterResources for
this and attempted to set the logging levels to pick up everything
(i.e. ALL), but if there's a way to expose more, I'm not aware of it.
*Have you debugged the execution and followed the counter() calls
all the way to the reporter?*
With the debugger, I traced one of the counter initializations and it
seems that no reporters were being found within the register call in
the MetricsRegistryImpl (i.e. this.reporters has no registered reporters):
if (this.reporters !=null) {
for(int i =0; i <this.reporters.size(); ++i) {
MetricRegistryImpl.ReporterAndSettings reporterAndSettings =
(MetricRegistryImpl.ReporterAndSettings)this.reporters.get(i);
try {
if (reporterAndSettings !=null) {
FrontMetricGroup front =new
FrontMetricGroup(reporterAndSettings.getSettings(),group);
reporterAndSettings.getReporter().notifyOfAddedMetric(metric,metricName, front);
}
}catch (Exception var11) {
LOG.warn("Error while registering metric: {}.",metricName,var11);
}
}
}
Perhaps this is an error on my part as I had assumed the following
would be sufficient to register my reporter (within a local /
minicluster environment):
private val metricsConfiguration =Configuration.fromMap(mutableMapOf(
ConfigConstants.METRICS_REPORTER_PREFIX +
"MockCustomMetricsReporter." +
ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX to
MockCustomMetricsReporter::class.java.name ))
@ClassRule @JvmField val flink =MiniClusterResource(
MiniClusterResourceConfiguration.Builder()
.setConfiguration(metricsConfiguration)
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(1)
.build()
)
However, it's clearly being recognized for the built-in metrics, just
not these custom ones that are being registered as they are triggering
the notifyOfAddedMetric() function within the reporter itself.
*Do you only see JobManager metrics, or is there somewhere also
something about the TaskManager?*
It looks like there are metrics coming from both the JobManager and
TaskManagers from the following examples that were coming out:
localhost.jobmanager.numRegisteredTaskManagers
.taskmanager.ad5aaade-cd54-44f4-a099-a765b53f79b4.Status.Shuffle.Netty.UsedMemorySegments
.taskmanager.ad5aaade-cd54-44f4-a099-a765b53f79b4.Status.JVM.Memory.Metaspace.Committed
localhost.jobmanager.Status.JVM.Memory.Direct.Count
I do agree that a factory implementation with a static reporter would
likely be a better approach, so I may explore that a bit more. As well
as adding some changes to the existing, albeit ghetto, implementation
for handling the dynamic metrics. I did see several references to a
MetricRegistry class, however I wasn't sure if that was the most
appropriate place to add this type of functionality or if it was
needed at all.
Thanks much,
Rion
On Tue, Mar 16, 2021 at 4:45 AM Chesnay Schepler <ches...@apache.org
<mailto:ches...@apache.org>> wrote:
Was there anything in the logs (ideally on debug)?
Have you debugged the execution and followed the counter() calls
all the way to the reporter?
Do you only see JobManager metrics, or is there somewhere also
something about the TaskManager?
I can see several issues with your code, but none that would fully
explain the issue:
a) your reporter is not thread-safe
b) you only differentiate metrics by name, which will lead to
quite a few collisions.
Be also aware that there will be 2 reporter instances; one for the
JM and one for the TM.
To remedy this, I would recommend creating a factory that returns
a static reporter instance instead; overall this tends to be cleaner.
Alternatively, when using the testing harnesses IIRC you can also
set set a custom MetricGroup implementation.
On 3/16/2021 4:13 AM, Rion Williams wrote:
Hi all,
Recently, I was working on adding some custom metrics to a Flink
job that required the use of dynamic labels (i.e. capturing
various counters that were "slicable" by things like tenant /
source, etc.).
I ended up handling it in a very naive fashion that would just
keep a dictionary of metrics that had already been registered and
update them accordingly which looked something like this:
class MyCustomProcessFunction:ProcessFunction<Event,Unit>() {
private lateinit var metrics:CustomMetricsRegistry override fun
open(parameters:Configuration) {
metrics =CustomMetricsRegistry(runtimeContext.metricGroup)
}
override fun
processElement(event:Event,context:Context,collector:Collector<Unit>) {
// Insert calls like metrics.inc("tenant-name", 4) here }
}
class CustomMetricsRegistry(private val
metricGroup:MetricGroup):Serializable {
// Increments a given metric by key fun
inc(metric:String,tenant:String,amount:Long =1) {
// Store a key for the metric val key ="$metric-$tenant" //
Store/register the metric if (!registeredMetrics.containsKey(key)){
registeredMetrics[key] =metricGroup .addGroup("tenant",tenant)
.counter(metric)
}
// Update the metric by a given amount
registeredMetrics[key]!!.inc(amount)
}
companion object {
private var registeredMetrics:HashMap<String,Counter> = hashMapOf()
}
}
Basically registering and updating new metrics for tenants as
they are encountered, which I've seen being emitted as expected
via hitting the appropriately configured metrics endpoint (using
a PrometheusReporter).
However, while I was trying to write a few unit tests for this, I
seemed to encounter an issue. I was following a Stack Overflow
post that was answered by @Chesnay Schepler
<mailto:ches...@apache.org> [0] that described the use of an
in-memory/embedded Flink cluster and a custom reporter that would
statically expose the underlying metrics.
So I took a shot at implementing something similar as follows:
*Flink Cluster Definition*
private val metricsConfiguration =Configuration.fromMap(mutableMapOf(
ConfigConstants.METRICS_REPORTER_PREFIX +
"MockCustomMetricsReporter." +
ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX to
MockCustomMetricsReporter::class.java.name ))
@ClassRule @JvmField val flinkCluster =MiniClusterResource(
MiniClusterResourceConfiguration.Builder()
.setConfiguration(metricsConfiguration)
.setNumberTaskManagers(1)
.setNumberSlotsPerTaskManager(1)
.build()
)
*Custom Reporter*
class MockCustomMetricsReporter :MetricReporter {
override fun open(metricConfig:MetricConfig) {}
override fun close() {}
override fun
notifyOfAddedMetric(metric:Metric,name:String,metricGroup:MetricGroup) {
// Store the metrics that are being registered as we see them if
(!registeredCustomMetrics.containsKey(name)){
registeredCustomMetrics[name] =metric }
}
override fun
notifyOfRemovedMetric(metric:Metric,name:String,metricGroup:MetricGroup) {
// Do nothing here }
companion object {
// Static reference to metrics as they are registered var
registeredCustomMetrics =HashMap<String,Metric>()
}
}
*Example Test*
@Test fun `Example Metrics Use Case`(){
// Arrange val stream
=StreamExecutionEnvironment.getExecutionEnvironment()
val events =listOf(
eventWithUsers("tenant1","us...@testing.com
<mailto:us...@testing.com>"),
eventWithUsers("tenant2","us...@testing.com
<mailto:us...@testing.com>"),
)
// Act stream
.fromCollection(events)
.process(MyCustomProcessFunction())
// Assert stream.execute()
assertTrue(MockCustomMetricsReporter.registeredCustomMetrics.size > 0)
}
While this test will pass, *the problem is that the custom
metrics defined dynamically (via the CustomMetricsRegistry
implementation) do not appear within the registeredCustomMetrics
collection*. In fact, there are 21 metrics that get registered
but all of them appear to be classic out-of-the-box metrics such
as CPU usage, number of task managers, load, various other Netty
and JVM stats, but no custom metrics are included.
I've tried multiple different configurations, implementations via
a custom TestHarness, etc. but for some reason the custom metrics
being defined are never triggering the notifyOfAddedMetric
function which would be responsible for adding them to the static
collection to be asserted against.
Any ideas / guidance would be more than welcome. Perhaps a
different approach? Based off examples I've encountered, the code
seems like it should "just work".
Thanks much,
Rion
[0] :
https://stackoverflow.com/questions/51675597/how-to-unitest-gauge-metrics-in-flink
<https://stackoverflow.com/questions/51675597/how-to-unitest-gauge-metrics-in-flink>
