Refactoring event logs

incubator/catalog-schema-registry.spec/src/main/resources/META-INF/zilla/schema_registry.idl

@@ -21,16 +21,16 @@ scope schema_registry
             REMOTE_ACCESS_REJECTED (1)
         }
 
-        struct SchemaRegistryRemoteAccessRejected extends core::event::Event
+        struct SchemaRegistryRemoteAccessRejectedEx extends core::stream::Extension
         {
             string8 method;
             string16 url;
             int16 status;
         }
 
-        union SchemaRegistryEvent switch (SchemaRegistryEventType)
+        union SchemaRegistryEventEx switch (SchemaRegistryEventType)
         {
-            case REMOTE_ACCESS_REJECTED: SchemaRegistryRemoteAccessRejected remoteAccessRejected;
+            case REMOTE_ACCESS_REJECTED: SchemaRegistryRemoteAccessRejectedEx remoteAccessRejected;
         }
     }
 }

incubator/exporter-stdout.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/client.yaml → incubator/catalog-schema-registry.spec/src/main/scripts/io/aklivity/zilla/specs/catalog/schema/registry/config/event.yaml

@@ -17,10 +17,22 @@
 name: test
 telemetry:
   exporters:
-    stdout0:
-      type: stdout
+    exporter0:
+      type: test
+      options:
+        events:
+          - qname: test.catalog0
+            message: REMOTE_ACCESS_REJECTED GET http://localhost:8081/schemas/ids/0 0
+catalogs:
+  catalog0:
+    type: schema-registry
+    options:
+      url: http://localhost:8081
 bindings:
-  app0:
-    type: kafka
-    kind: client
-    exit: net0
+  net0:
+    type: test
+    kind: server
+    options:
+      catalogs:
+        - catalog0
+    exit: app0

incubator/catalog-schema-registry/pom.xml

@@ -24,7 +24,7 @@
   <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.80</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.90</jacoco.coverage.ratio>
     <jacoco.missed.count>0</jacoco.missed.count>
   </properties>
 

incubator/catalog-schema-registry/src/main/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/SchemaRegistryEventContext.java

@@ -14,23 +14,28 @@
  */
 package io.aklivity.zilla.runtime.catalog.schema.registry.internal;
 
+import static io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.SchemaRegistryEventType.REMOTE_ACCESS_REJECTED;
+
 import java.net.http.HttpRequest;
 import java.nio.ByteBuffer;
 import java.time.Clock;
 
-import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.AtomicBuffer;
 import org.agrona.concurrent.UnsafeBuffer;
 
-import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.SchemaRegistryEventFW;
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.EventFW;
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.SchemaRegistryEventExFW;
 import io.aklivity.zilla.runtime.engine.EngineContext;
 import io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer;
 
 public class SchemaRegistryEventContext
 {
     private static final int EVENT_BUFFER_CAPACITY = 1024;
 
-    private final SchemaRegistryEventFW.Builder schemaRegistryEventRW = new SchemaRegistryEventFW.Builder();
-    private final MutableDirectBuffer eventBuffer = new UnsafeBuffer(ByteBuffer.allocate(EVENT_BUFFER_CAPACITY));
+    private final AtomicBuffer eventBuffer = new UnsafeBuffer(ByteBuffer.allocate(EVENT_BUFFER_CAPACITY));
+    private final AtomicBuffer extensionBuffer = new UnsafeBuffer(ByteBuffer.allocate(EVENT_BUFFER_CAPACITY));
+    private final EventFW.Builder eventRW = new EventFW.Builder();
+    private final SchemaRegistryEventExFW.Builder schemaRegistryEventExRW = new SchemaRegistryEventExFW.Builder();
     private final int schemaRegistryTypeId;
     private final MessageConsumer eventWriter;
     private final Clock clock;
@@ -48,17 +53,22 @@ public void remoteAccessRejected(
         HttpRequest httpRequest,
         int status)
     {
-        SchemaRegistryEventFW event = schemaRegistryEventRW
-            .wrap(eventBuffer, 0, eventBuffer.capacity())
+        SchemaRegistryEventExFW extension = schemaRegistryEventExRW
+            .wrap(extensionBuffer, 0, extensionBuffer.capacity())
             .remoteAccessRejected(e -> e
-                .timestamp(clock.millis())
-                .traceId(0L)
-                .namespacedId(catalogId)
+                .typeId(REMOTE_ACCESS_REJECTED.value())
                 .method(httpRequest.method())
                 .url(httpRequest.uri().toString())
                 .status((short) status)
             )
             .build();
+        EventFW event = eventRW
+            .wrap(eventBuffer, 0, eventBuffer.capacity())
+            .timestamp(clock.millis())
+            .traceId(0L)
+            .namespacedId(catalogId)
+            .extension(extension.buffer(), extension.offset(), extension.limit())
+            .build();
         eventWriter.accept(schemaRegistryTypeId, event.buffer(), event.offset(), event.limit());
     }
 }

incubator/catalog-schema-registry/src/main/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/SchemaRegistryEventFormatter.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2021-2023 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.catalog.schema.registry.internal;
+
+import org.agrona.DirectBuffer;
+
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.StringFW;
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.EventFW;
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.SchemaRegistryEventExFW;
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.event.SchemaRegistryRemoteAccessRejectedExFW;
+import io.aklivity.zilla.runtime.engine.Configuration;
+import io.aklivity.zilla.runtime.engine.event.EventFormatterSpi;
+
+public final class SchemaRegistryEventFormatter implements EventFormatterSpi
+{
+    private static final String REMOTE_ACCESS_REJECTED = "REMOTE_ACCESS_REJECTED %s %s %d";
+
+    private final EventFW eventRO = new EventFW();
+    private final SchemaRegistryEventExFW schemaRegistryEventExRO = new SchemaRegistryEventExFW();
+
+    SchemaRegistryEventFormatter(
+        Configuration config)
+    {
+    }
+
+    public String format(
+        DirectBuffer buffer,
+        int index,
+        int length)
+    {
+        final EventFW event = eventRO.wrap(buffer, index, index + length);
+        final SchemaRegistryEventExFW extension = schemaRegistryEventExRO
+            .wrap(event.extension().buffer(), event.extension().offset(), event.extension().limit());
+        String result = null;
+        switch (extension.kind())
+        {
+        case REMOTE_ACCESS_REJECTED:
+        {
+            SchemaRegistryRemoteAccessRejectedExFW ex = extension.remoteAccessRejected();
+            result = String.format(REMOTE_ACCESS_REJECTED, asString(ex.method()), asString(ex.url()),
+                ex.status());
+            break;
+        }
+        }
+        return result;
+    }
+
+    private static String asString(
+        StringFW stringFW)
+    {
+        String s = stringFW.asString();
+        return s == null ? "" : s;
+    }
+}

incubator/catalog-schema-registry/src/main/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/SchemaRegistryEventFormatterFactory.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2021-2023 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.catalog.schema.registry.internal;
+
+import io.aklivity.zilla.runtime.engine.Configuration;
+import io.aklivity.zilla.runtime.engine.event.EventFormatterFactorySpi;
+
+public final class SchemaRegistryEventFormatterFactory implements EventFormatterFactorySpi
+{
+    @Override
+    public SchemaRegistryEventFormatter create(
+        Configuration config)
+    {
+        return new SchemaRegistryEventFormatter(config);
+    }
+
+    @Override
+    public String type()
+    {
+        return SchemaRegistryCatalog.NAME;
+    }
+}

incubator/catalog-schema-registry/src/main/moditect/module-info.java

@@ -22,4 +22,7 @@
 
     provides io.aklivity.zilla.runtime.engine.config.OptionsConfigAdapterSpi
         with io.aklivity.zilla.runtime.catalog.schema.registry.internal.config.SchemaRegistryOptionsConfigAdapter;
+
+    provides io.aklivity.zilla.runtime.engine.event.EventFormatterFactorySpi
+        with io.aklivity.zilla.runtime.catalog.schema.registry.internal.SchemaRegistryEventFormatterFactory;
 }

incubator/catalog-schema-registry/src/main/resources/META-INF/services/io.aklivity.zilla.runtime.engine.event.EventFormatterFactorySpi

@@ -0,0 +1 @@
+io.aklivity.zilla.runtime.catalog.schema.registry.internal.SchemaRegistryEventFormatterFactory

incubator/catalog-schema-registry/src/test/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/EventIT.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2021-2023 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.catalog.schema.registry.internal;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.rules.RuleChain.outerRule;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.DisableOnDebug;
+import org.junit.rules.TestRule;
+import org.junit.rules.Timeout;
+import org.kaazing.k3po.junit.annotation.Specification;
+import org.kaazing.k3po.junit.rules.K3poRule;
+
+import io.aklivity.zilla.runtime.engine.test.EngineRule;
+import io.aklivity.zilla.runtime.engine.test.annotation.Configuration;
+
+public class EventIT
+{
+    private final K3poRule k3po = new K3poRule()
+        .addScriptRoot("net", "io/aklivity/zilla/specs/engine/streams/network")
+        .addScriptRoot("app", "io/aklivity/zilla/specs/engine/streams/application");
+
+    private final TestRule timeout = new DisableOnDebug(new Timeout(10, SECONDS));
+
+    private final EngineRule engine = new EngineRule()
+        .directory("target/zilla-itests")
+        .countersBufferCapacity(4096)
+        .configurationRoot("io/aklivity/zilla/specs/catalog/schema/registry/config")
+        .external("app0")
+        .clean();
+
+    @Rule
+    public final TestRule chain = outerRule(engine).around(k3po).around(timeout);
+
+    @Test
+    @Configuration("event.yaml")
+    @Specification({
+        "${net}/event/client",
+        "${app}/event/server"
+    })
+    public void shouldLogEvents() throws Exception
+    {
+        k3po.finish();
+    }
+}

incubator/exporter-stdout.spec/pom.xml

@@ -41,18 +41,6 @@
       <artifactId>engine.spec</artifactId>
       <version>${project.version}</version>
     </dependency>
-    <dependency>
-      <groupId>${project.groupId}</groupId>
-      <artifactId>binding-http.spec</artifactId>
-      <version>${project.version}</version>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>${project.groupId}</groupId>
-      <artifactId>binding-proxy.spec</artifactId>
-      <version>${project.version}</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>