Add http.active.requests and http.duration metrics

incubator/exporter-prometheus/src/main/java/io/aklivity/zilla/runtime/exporter/prometheus/internal/descriptor/PrometheusMetricDescriptor.java

@@ -16,6 +16,7 @@
 
 import static io.aklivity.zilla.runtime.engine.metrics.Metric.Kind.COUNTER;
 import static io.aklivity.zilla.runtime.engine.metrics.Metric.Unit.BYTES;
+import static io.aklivity.zilla.runtime.engine.metrics.Metric.Unit.NANOSECONDS;
 
 import java.util.Map;
 import java.util.function.Function;
@@ -57,21 +58,27 @@ public String kind(
     public String name(
         String internalName)
     {
-        String result = names.get(internalName);
-        if (result == null)
+        return names.computeIfAbsent(internalName, this::externalName);
+    }
+
+    private String externalName(
+        String internalName)
+    {
+        String result;
+        Metric metric = metricResolver.apply(internalName);
+        result = metric.name();
+        result = result.replace('.', '_');
+        if (metric.unit() == BYTES)
+        {
+            result += "_bytes";
+        }
+        else if (metric.unit() == NANOSECONDS)
+        {
+            result += "_nanoseconds";
+        }
+        if (metric.kind() == COUNTER)
         {
-            Metric metric = metricResolver.apply(internalName);
-            result = metric.name();
-            result = result.replace('.', '_');
-            if (metric.unit() == BYTES)
-            {
-                result += "_bytes";
-            }
-            if (metric.kind() == COUNTER)
-            {
-                result += "_total";
-            }
-            names.put(internalName, result);
+            result += "_total";
         }
         return result;
     }

incubator/metrics-http.spec/src/main/scripts/io/aklivity/zilla/specs/metrics/http/schema/http.schema.patch.json

@@ -8,5 +8,15 @@
     "op": "add",
     "path": "/$defs/telemetry/metrics/items/enum/-",
     "value": "http.response.size"
+  },
+  {
+    "op": "add",
+    "path": "/$defs/telemetry/metrics/items/enum/-",
+    "value": "http.active.requests"
+  },
+  {
+    "op": "add",
+    "path": "/$defs/telemetry/metrics/items/enum/-",
+    "value": "http.duration"
   }
 ]

incubator/metrics-http/src/main/java/io/aklivity/zilla/runtime/metrics/http/internal/HttpActiveRequestsMetric.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2021-2022 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.metrics.http.internal;
+
+import io.aklivity.zilla.runtime.engine.EngineContext;
+import io.aklivity.zilla.runtime.engine.metrics.Metric;
+import io.aklivity.zilla.runtime.engine.metrics.MetricContext;
+
+public class HttpActiveRequestsMetric implements Metric
+{
+    private static final String GROUP = HttpMetricGroup.NAME;
+    private static final String NAME = String.format("%s.%s", GROUP, "active.requests");
+    private static final String DESCRIPTION = "Number of active HTTP requests";
+
+    @Override
+    public String name()
+    {
+        return NAME;
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.GAUGE;
+    }
+
+    @Override
+    public Unit unit()
+    {
+        return Unit.COUNT;
+    }
+
+    @Override
+    public String description()
+    {
+        return DESCRIPTION;
+    }
+
+    @Override
+    public MetricContext supply(
+        EngineContext context)
+    {
+        return new HttpActiveRequestsMetricContext(GROUP, kind());
+    }
+}

incubator/metrics-http/src/main/java/io/aklivity/zilla/runtime/metrics/http/internal/HttpActiveRequestsMetricContext.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright 2021-2022 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.metrics.http.internal;
+
+import static io.aklivity.zilla.runtime.engine.metrics.MetricContext.Direction.BOTH;
+import static io.aklivity.zilla.runtime.metrics.http.internal.HttpUtils.initialId;
+
+import java.util.function.LongConsumer;
+
+import org.agrona.DirectBuffer;
+import org.agrona.collections.Long2LongHashMap;
+
+import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
+import io.aklivity.zilla.runtime.engine.metrics.Metric;
+import io.aklivity.zilla.runtime.engine.metrics.MetricContext;
+import io.aklivity.zilla.runtime.metrics.http.internal.types.stream.AbortFW;
+import io.aklivity.zilla.runtime.metrics.http.internal.types.stream.BeginFW;
+import io.aklivity.zilla.runtime.metrics.http.internal.types.stream.EndFW;
+import io.aklivity.zilla.runtime.metrics.http.internal.types.stream.FrameFW;
+import io.aklivity.zilla.runtime.metrics.http.internal.types.stream.ResetFW;
+
+public final class HttpActiveRequestsMetricContext implements MetricContext
+{
+    private final String group;
+    private final Metric.Kind kind;
+    private final FrameFW frameRO = new FrameFW();
+
+    public HttpActiveRequestsMetricContext(
+        String group,
+        Metric.Kind kind)
+    {
+        this.group = group;
+        this.kind = kind;
+    }
+
+    @Override
+    public String group()
+    {
+        return group;
+    }
+
+    @Override
+    public Metric.Kind kind()
+    {
+        return kind;
+    }
+
+    @Override
+    public Direction direction()
+    {
+        return BOTH;
+    }
+
+    @Override
+    public MessageConsumer supply(
+        LongConsumer recorder)
+    {
+        return new HttpActiveRequestsMetricHandler(recorder);
+    }
+
+    private final class HttpActiveRequestsMetricHandler implements MessageConsumer
+    {
+        private static final long INITIAL_VALUE = 0L;
+        private static final long EXCHANGE_CLOSED = 0b11L;
+
+        private final LongConsumer recorder;
+        private final Long2LongHashMap exchanges;
+
+        private HttpActiveRequestsMetricHandler(
+            LongConsumer recorder)
+        {
+            this.recorder = recorder;
+            this.exchanges = new Long2LongHashMap(INITIAL_VALUE);
+        }
+
+        @Override
+        public void accept(
+            int msgTypeId,
+            DirectBuffer buffer,
+            int index,
+            int length)
+        {
+            final FrameFW frame = frameRO.wrap(buffer, index, index + length);
+            final long streamId = frame.streamId();
+            final long exchangeId = initialId(streamId);
+            long direction = HttpUtils.direction(streamId);
+            switch (msgTypeId)
+            {
+            case BeginFW.TYPE_ID:
+                if (direction == 1L) //received
+                {
+                    recorder.accept(1L);
+                }
+                break;
+            case ResetFW.TYPE_ID:
+            case AbortFW.TYPE_ID:
+            case EndFW.TYPE_ID:
+                final long mask = 1L << direction;
+                final long status = exchanges.get(exchangeId) | mask; // mark current direction as closed
+                if (status == EXCHANGE_CLOSED) // both received and sent streams are closed
+                {
+                    exchanges.remove(exchangeId);
+                    recorder.accept(-1L);
+                }
+                else
+                {
+                    exchanges.put(exchangeId, status);
+                }
+                break;
+            }
+        }
+    }
+}

incubator/metrics-http/src/main/java/io/aklivity/zilla/runtime/metrics/http/internal/HttpDurationMetric.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2021-2022 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.metrics.http.internal;
+
+import io.aklivity.zilla.runtime.engine.EngineContext;
+import io.aklivity.zilla.runtime.engine.metrics.Metric;
+import io.aklivity.zilla.runtime.engine.metrics.MetricContext;
+
+public class HttpDurationMetric implements Metric
+{
+    private static final String GROUP = HttpMetricGroup.NAME;
+    private static final String NAME = String.format("%s.%s", GROUP, "duration");
+    private static final String DESCRIPTION = "Duration of HTTP requests";
+
+    @Override
+    public String name()
+    {
+        return NAME;
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.HISTOGRAM;
+    }
+
+    @Override
+    public Unit unit()
+    {
+        return Unit.NANOSECONDS;
+    }
+
+    @Override
+    public String description()
+    {
+        return DESCRIPTION;
+    }
+
+    @Override
+    public MetricContext supply(
+        EngineContext context)
+    {
+        return new HttpDurationMetricContext(GROUP, kind());
+    }
+}