merge: #9371 #9373

9371: Configure record filter for metrics exporter r=korthout a=korthout

## Description

<!-- Please explain the changes you made here. -->

Configures a record filter for the Metrics exporter.

## Related issues

<!-- Which issues are closed by this PR or are related -->

closes #9240 
relates to #6442 



9373: deps(go): bump github.com/docker/docker from 20.10.15+incompatible to 20.10.16+incompatible in /clients/go r=Zelldon a=dependabot[bot]

Bumps [github.com/docker/docker](https://github.com/docker/docker) from 20.10.15+incompatible to 20.10.16+incompatible.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/docker/docker/releases">github.com/docker/docker's releases</a>.</em></p>
<blockquote>
<h2>v20.10.16</h2>
<p>This release of Docker Engine fixes a regression in the Docker CLI builds for
macOS, fixes an issue with <code>docker stats</code> when using containerd 1.5 and up,
and updates the Go runtime to include a fix for <a href="http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2022-29526">CVE-2022-29526</a>.</p>
<h3>Client</h3>
<ul>
<li>Fix a regression in binaries for macOS introduced in <a href="%5B#201015%5D(https://github-redirect.dependabot.com/docker/docker/issues/201015)">20.10.15</a>, which
resulted in a panic <a href="https://github-redirect.dependabot.com/docker/cli/pull/3592">docker/cli#43426</a>.</li>
<li>Update golang.org/x/sys dependency which contains a fix for
<a href="http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2022-29526">CVE-2022-29526</a>.</li>
</ul>
<h3>Daemon</h3>
<ul>
<li>Fix an issue where <code>docker stats</code> was showing empty stats when running with
containerd 1.5.0 or up <a href="https://github-redirect.dependabot.com/moby/moby/pull/43567">moby/moby#43567</a>.</li>
<li>Update the <code>golang.org/x/sys</code> build-time dependency which contains a fix for <a href="http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2022-29526">CVE-2022-29526</a>.</li>
</ul>
<h3>Packaging</h3>
<ul>
<li>Update Go runtime to <a href="https://go.dev/doc/devel/release#go1.17.minor">1.17.10</a>,
which contains a fix for <a href="http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2022-29526">CVE-2022-29526</a>.</li>
<li>Use &quot;weak&quot; dependencies for the <code>docker scan</code> CLI plugin, to prevent a
&quot;conflicting requests&quot; error when users performed an off-line installation from
downloaded RPM packages <a href="https://github-redirect.dependabot.com/docker/docker-ce-packaging/pull/659">docker/docker-ce-packaging#659</a>.</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/moby/moby/commit/f756502055d2e36a84f2068e6620bea5ecf09058"><code>f756502</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/docker/issues/43582">#43582</a> from thaJeztah/20.10_bump_golang_1.17.10</li>
<li><a href="https://github.com/moby/moby/commit/a15acb4bd6b9b65f93de5f52de56a29ae4d63a4c"><code>a15acb4</code></a> [20.10] vendor: golang.org/x/sys v0.0.0-20220412211240-33da011f77ad</li>
<li><a href="https://github.com/moby/moby/commit/5f2e0b79adc4d1786043176ee599e73f33430989"><code>5f2e0b7</code></a> [20.10] update golang to 1.17.10</li>
<li><a href="https://github.com/moby/moby/commit/462cd7de50edbba462ca2de82ceb10d26ac1e6bb"><code>462cd7d</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/docker/docker/issues/43567">#43567</a> from 42wim/fixstats</li>
<li><a href="https://github.com/moby/moby/commit/be7855fdbe3136f144feb3b9340e7d21c5c45e1e"><code>be7855f</code></a> vendor: update github.com/containerd/cgroups and github.com/cilium/ebpf</li>
<li>See full diff in <a href="https://github.com/docker/docker/compare/v20.10.15...v20.10.16">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=github.com/docker/docker&package-manager=go_modules&previous-version=20.10.15+incompatible&new-version=20.10.16+incompatible)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting ``@dependabot` rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- ``@dependabot` rebase` will rebase this PR
- ``@dependabot` recreate` will recreate this PR, overwriting any edits that have been made to it
- ``@dependabot` merge` will merge this PR after your CI passes on it
- ``@dependabot` squash and merge` will squash and merge this PR after your CI passes on it
- ``@dependabot` cancel merge` will cancel a previously requested merge and block automerging
- ``@dependabot` reopen` will reopen this PR if it is closed
- ``@dependabot` close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- ``@dependabot` ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- ``@dependabot` ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- ``@dependabot` ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

Co-authored-by: Nico Korthout <nico.korthout@camunda.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.gitignore

@@ -30,6 +30,7 @@ faces-config.NavData
 *.lock.db
 .cache-main
 .cache-tests
+.jqwik-database
 
 *.DS_Store
 .java-version

broker/pom.xml

@@ -203,6 +203,12 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.camunda</groupId>
+      <artifactId>zeebe-exporter-test</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>io.camunda</groupId>
       <artifactId>zeebe-workflow-engine</artifactId>

broker/src/main/java/io/camunda/zeebe/broker/exporter/metrics/ExecutionLatencyMetrics.java

@@ -56,6 +56,10 @@ public void observeJobActivationTime(
         .observe(latencyInSeconds(creationTimeMs, activationTimeMs));
   }
 
+  public Histogram getJobLifeTime() {
+    return JOB_LIFE_TIME;
+  }
+
   /**
    * Takes start and end time in milliseconds and calculates the difference (latency) in seconds.
    *

broker/src/main/java/io/camunda/zeebe/broker/exporter/metrics/MetricsExporter.java

@@ -9,6 +9,8 @@
 
 import io.camunda.zeebe.broker.system.configuration.ExporterCfg;
 import io.camunda.zeebe.exporter.api.Exporter;
+import io.camunda.zeebe.exporter.api.context.Context;
+import io.camunda.zeebe.exporter.api.context.Context.RecordFilter;
 import io.camunda.zeebe.exporter.api.context.Controller;
 import io.camunda.zeebe.protocol.record.Record;
 import io.camunda.zeebe.protocol.record.RecordType;
@@ -21,6 +23,7 @@
 import io.camunda.zeebe.protocol.record.value.ProcessInstanceRecordValue;
 import java.time.Duration;
 import java.util.NavigableMap;
+import java.util.Set;
 import java.util.TreeMap;
 import org.agrona.collections.Long2LongHashMap;
 
@@ -39,13 +42,36 @@ public class MetricsExporter implements Exporter {
   private Controller controller;
 
   public MetricsExporter() {
-    executionLatencyMetrics = new ExecutionLatencyMetrics();
+    this(new ExecutionLatencyMetrics());
+  }
+
+  public MetricsExporter(final ExecutionLatencyMetrics executionLatencyMetrics) {
+    this.executionLatencyMetrics = executionLatencyMetrics;
     jobKeyToCreationTimeMap = new Long2LongHashMap(-1);
     processInstanceKeyToCreationTimeMap = new Long2LongHashMap(-1);
     creationTimeToJobKeyNavigableMap = new TreeMap<>();
     creationTimeToProcessInstanceKeyNavigableMap = new TreeMap<>();
   }
 
+  @Override
+  public void configure(final Context context) throws Exception {
+    context.setFilter(
+        new RecordFilter() {
+          private static final Set<ValueType> ACCEPTED_VALUE_TYPES =
+              Set.of(ValueType.JOB, ValueType.JOB_BATCH, ValueType.PROCESS_INSTANCE);
+
+          @Override
+          public boolean acceptType(final RecordType recordType) {
+            return recordType == RecordType.EVENT;
+          }
+
+          @Override
+          public boolean acceptValue(final ValueType valueType) {
+            return ACCEPTED_VALUE_TYPES.contains(valueType);
+          }
+        });
+  }
+
   @Override
   public void open(final Controller controller) {
     this.controller = controller;

broker/src/test/java/io/camunda/zeebe/broker/exporter/metrics/MetricsExporterTest.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.broker.exporter.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.camunda.zeebe.exporter.test.ExporterTestContext;
+import io.camunda.zeebe.exporter.test.ExporterTestController;
+import io.camunda.zeebe.protocol.Protocol;
+import io.camunda.zeebe.protocol.record.ImmutableRecord;
+import io.camunda.zeebe.protocol.record.RecordType;
+import io.camunda.zeebe.protocol.record.ValueType;
+import io.camunda.zeebe.protocol.record.intent.JobIntent;
+import java.util.Arrays;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class MetricsExporterTest {
+
+  @Test
+  void shouldObserveJobLifetime() {
+    // given
+    final var metrics = new ExecutionLatencyMetrics();
+    final var exporter = new MetricsExporter(metrics);
+    exporter.open(new ExporterTestController());
+    assertThat(metrics.getJobLifeTime().collect())
+        .flatMap(x -> x.samples)
+        .describedAs("Expected no metrics to be recorded at start of test")
+        .isEmpty();
+
+    // when
+    exporter.export(
+        ImmutableRecord.builder()
+            .withRecordType(RecordType.EVENT)
+            .withValueType(ValueType.JOB)
+            .withIntent(JobIntent.CREATED)
+            .withTimestamp(1651505728460L)
+            .withKey(Protocol.encodePartitionId(1, 1))
+            .build());
+    exporter.export(
+        ImmutableRecord.builder()
+            .withRecordType(RecordType.EVENT)
+            .withValueType(ValueType.JOB)
+            .withIntent(JobIntent.COMPLETED)
+            .withTimestamp(1651505729571L)
+            .withKey(Protocol.encodePartitionId(1, 1))
+            .build());
+
+    // then
+    assertThat(metrics.getJobLifeTime().collect())
+        .flatMap(x -> x.samples)
+        .filteredOn(s -> s.name.equals("zeebe_job_life_time_count"))
+        .map(s -> s.value)
+        .describedAs("Expected exactly 1 observed job_life_time sample counted")
+        .containsExactly(1d);
+  }
+
+  @Nested
+  @DisplayName("MetricsExporter should configure a Filter")
+  class FilterTest {
+
+    static Stream<TypeCombination> acceptedCombinations() {
+      return Stream.of(
+          new TypeCombination(RecordType.EVENT, ValueType.JOB),
+          new TypeCombination(RecordType.EVENT, ValueType.JOB_BATCH),
+          new TypeCombination(RecordType.EVENT, ValueType.PROCESS_INSTANCE));
+    }
+
+    /** Returns the inverse of {@link #acceptedCombinations()}. */
+    static Stream<TypeCombination> rejectedCombinations() {
+      return allCombinations().filter(any -> acceptedCombinations().noneMatch(any::equals));
+    }
+
+    static Stream<TypeCombination> allCombinations() {
+      return Arrays.stream(RecordType.values())
+          .flatMap(
+              recordType ->
+                  Arrays.stream(ValueType.values())
+                      .map(valueType -> new TypeCombination(recordType, valueType)));
+    }
+
+    @ParameterizedTest
+    @DisplayName("accepting records of specific RecordType and ValueType")
+    @MethodSource("acceptedCombinations")
+    void shouldConfigureFilterAccepting(final TypeCombination combination) throws Exception {
+      // given
+      final var recordType = combination.recordType();
+      final var valueType = combination.valueType();
+      final var context = new ExporterTestContext();
+
+      // when
+      new MetricsExporter().configure(context);
+
+      // then
+      final var recordFilter = context.getRecordFilter();
+      assertThat(recordFilter.acceptType(recordType) && recordFilter.acceptValue(valueType))
+          .describedAs(
+              "Expect RecordFilter to accept record of RecordType %s and ValueType %s",
+              recordType, valueType)
+          .isTrue();
+    }
+
+    @ParameterizedTest
+    @DisplayName("rejecting records of specific RecordType and ValueType")
+    @MethodSource("rejectedCombinations")
+    void shouldConfigureFilterRejecting(final TypeCombination combination) throws Exception {
+      // given
+      final var recordType = combination.recordType();
+      final var valueType = combination.valueType();
+      final var context = new ExporterTestContext();
+
+      // when
+      new MetricsExporter().configure(context);
+
+      // then
+      final var recordFilter = context.getRecordFilter();
+      assertThat(recordFilter.acceptType(recordType) && recordFilter.acceptValue(valueType))
+          .describedAs(
+              "Expect RecordFilter to reject record of RecordType %s and ValueType %s",
+              recordType, valueType)
+          .isFalse();
+    }
+
+    /** Defines a combination of a RecordType and a ValueType. */
+    record TypeCombination(RecordType recordType, ValueType valueType) {}
+  }
+}

clients/go/go.mod

@@ -3,7 +3,7 @@ module github.com/camunda/zeebe/clients/go/v8
 go 1.15
 
 require (
-	github.com/docker/docker v20.10.15+incompatible
+	github.com/docker/docker v20.10.16+incompatible
 	github.com/docker/go-connections v0.4.0
 	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/go-ozzo/ozzo-validation/v4 v4.3.0

clients/go/go.sum

@@ -251,8 +251,8 @@ github.com/docker/distribution v2.7.1-0.20190205005809-0d3efadf0154+incompatible
 github.com/docker/distribution v2.7.1+incompatible h1:a5mlkVzth6W5A4fOsS3D2EO5BUmsJpcB+cRlLU7cSug=
 github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/docker v20.10.11+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
-github.com/docker/docker v20.10.15+incompatible h1:dk9FewY/9Xwm4ay/HViEEHSQuM/kL4F+JaG6GQdgmGo=
-github.com/docker/docker v20.10.15+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v20.10.16+incompatible h1:2Db6ZR/+FUR3hqPMwnogOPHFn405crbpxvWzKovETOQ=
+github.com/docker/docker v20.10.16+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ=
 github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
 github.com/docker/go-events v0.0.0-20170721190031-9461782956ad/go.mod h1:Uw6UezgYA44ePAFQYUehOuCzmy5zmg/+nl2ZfMWGkpA=

clients/go/vendor/modules.txt

@@ -44,7 +44,7 @@ github.com/davecgh/go-spew/spew
 # github.com/docker/distribution v2.7.1+incompatible
 github.com/docker/distribution/digestset
 github.com/docker/distribution/reference
-# github.com/docker/docker v20.10.15+incompatible
+# github.com/docker/docker v20.10.16+incompatible
 ## explicit
 github.com/docker/docker/api
 github.com/docker/docker/api/types