[Issue apache#11799][logging] Base implementation of structured loggi…

…ng (apache#11901)

Includes the most simple parts of slf4j structured logging. timed(),
sampled() and enum logging to follow.

buildtools/src/main/resources/log4j2.xml

@@ -22,7 +22,7 @@
 <Configuration status="INFO">
     <Appenders>
         <Console name="Console" target="SYSTEM_OUT">
-            <PatternLayout pattern="%d{HH:mm:ss.SSS} %-5level [%t{12}] %c{1.}@%L - %msg%n" />
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} %-5level [%t{12}] %c{1.}@%L - %msg %X%n" />
         </Console>
     </Appenders>
     <Loggers>

pom.xml

@@ -203,6 +203,8 @@ flexible messaging model and an intuitive client API.</description>
     <cassandra.version>3.6.0</cassandra.version>
     <disruptor.version>3.4.0</disruptor.version>
     <testcontainers.version>1.15.3</testcontainers.version>
+    <hamcrest.version>2.2</hamcrest.version>
+
     <!-- Set docker-java.version to the version of docker-java used in Testcontainers -->
     <docker-java.version>3.2.8</docker-java.version>
     <kerby.version>1.1.1</kerby.version>
@@ -277,6 +279,13 @@ flexible messaging model and an intuitive client API.</description>
         </exclusions>
       </dependency>
 
+      <dependency>
+        <groupId>org.hamcrest</groupId>
+        <artifactId>hamcrest</artifactId>
+        <version>${hamcrest.version}</version>
+        <scope>test</scope>
+      </dependency>
+
       <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>

structured-event-log/pom.xml

@@ -42,6 +42,11 @@
       <artifactId>testng</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>

structured-event-log/src/main/java/org/apache/pulsar/structuredeventlog/EventResourcesImpl.java

@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.structuredeventlog;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+
+public class EventResourcesImpl implements EventResources {
+    private final EventResourcesImpl parent;
+    private List<Object> resources = null;
+
+    public EventResourcesImpl(EventResourcesImpl parent) {
+        this.parent = parent;
+    }
+
+    @Override
+    public EventResources resource(String key, Object value) {
+        getResources().add(key);
+        getResources().add(value);
+        return this;
+    }
+
+    @Override
+    public EventResources resource(String key, Supplier<String> value) {
+        resource(key, (Object)value);
+        return this;
+    }
+
+    public void copyFrom(EventResourcesImpl other) {
+        // can't use forEach because we want to avoid toString at this point
+        List<Object> resources = getResources();
+        if (other.parent != null) {
+            copyFrom(other.parent);
+        }
+
+        if (other.resources != null) {
+            resources.addAll(other.resources);
+        }
+    }
+
+    public void forEach(BiConsumer<String, String> process) {
+        if (parent != null) {
+            parent.forEach(process);
+        }
+        if (resources != null) {
+            forEach(resources, process);
+        }
+    }
+
+    private List<Object> getResources() {
+        if (resources == null) {
+            resources = new ArrayList<>(2);
+        }
+        return resources;
+    }
+
+    public static void forEach(List<Object> list, BiConsumer<String, String> process) {
+        for (int i = 0; i < list.size() - 1; i += 2) {
+            String key = String.valueOf(list.get(i));
+            Object value = list.get(i + 1);
+            if (value instanceof Supplier) {
+                value = ((Supplier<?>)value).get();
+            }
+            process.accept(key, String.valueOf(value));
+        }
+    }
+}

structured-event-log/src/main/java/org/apache/pulsar/structuredeventlog/StructuredEventLog.java

@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.structuredeventlog;
 
+import org.apache.pulsar.structuredeventlog.slf4j.Slf4jStructuredEventLog;
+
 /**
  * Structured event logging interface
  *
@@ -84,6 +86,6 @@ public interface StructuredEventLog {
      * Create a new logger object, from which root events can be created.
      */
     public static StructuredEventLog newLogger() {
-        return null;
+        return Slf4jStructuredEventLog.INSTANCE;
     }
 }

structured-event-log/src/main/java/org/apache/pulsar/structuredeventlog/slf4j/Slf4jEvent.java

@@ -0,0 +1,206 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.structuredeventlog.slf4j;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import org.apache.pulsar.structuredeventlog.Event;
+import org.apache.pulsar.structuredeventlog.EventResources;
+import org.apache.pulsar.structuredeventlog.EventResourcesImpl;
+
+import org.slf4j.MDC;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class Slf4jEvent implements Event {
+    private final String id;
+    private String traceId = null;
+    private String parentId = null;
+    private List<Object> attributes = null;
+    private Level level = Level.INFO;
+    private Throwable throwable = null;
+    private final EventResourcesImpl resources;
+
+    Slf4jEvent(EventResourcesImpl parentResources) {
+        this.id = randomId();
+        this.resources = new EventResourcesImpl(parentResources);
+    }
+
+    @Override
+    public Event newChildEvent() {
+        return new Slf4jEvent(resources).traceId(traceId).parentId(id);
+    }
+
+    @Override
+    public Event traceId(String traceId) {
+        this.traceId = traceId;
+        return this;
+    }
+
+    @Override
+    public Event parentId(String parentId) {
+        this.parentId = parentId;
+        return this;
+    }
+
+    @Override
+    public Event timed() {
+        throw new UnsupportedOperationException("TODO");
+    }
+
+    @Override
+    public Event sampled(Object samplingKey, int duration, TimeUnit unit) {
+        throw new UnsupportedOperationException("TODO");
+    }
+
+    @Override
+    public Event resources(EventResources other) {
+        if (other instanceof EventResourcesImpl) {
+            this.resources.copyFrom((EventResourcesImpl)other);
+        }
+        return this;
+    }
+
+    @Override
+    public Event resource(String key, Object value) {
+        resources.resource(key, value);
+        return this;
+    }
+
+    @Override
+    public Event resource(String key, Supplier<String> value) {
+        resources.resource(key, value);
+        return this;
+    }
+
+    @Override
+    public Event attr(String key, Object value) {
+        getAttributes().add(key);
+        getAttributes().add(value);
+        return this;
+    }
+
+    @Override
+    public Event attr(String key, Supplier<String> value) {
+        this.attr(key, (Object)value);
+        return this;
+    }
+
+    @Override
+    public Event exception(Throwable t) {
+        this.throwable = t;
+        return this;
+    }
+
+    @Override
+    public Event atError() {
+        this.level = Level.ERROR;
+        return this;
+    }
+
+    @Override
+    public Event atInfo() {
+        this.level = Level.INFO;
+        return this;
+    }
+
+    @Override
+    public Event atWarn() {
+        this.level = Level.WARN;
+        return this;
+    }
+
+    @Override
+    public void log(Enum<?> event) {
+        throw new UnsupportedOperationException("TODO");
+    }
+
+    @Override
+    public void log(String event) {
+        try {
+            MDC.put("id", id);
+            if (traceId != null) {
+                MDC.put("traceId", traceId);
+            }
+            if (parentId != null) {
+                MDC.put("parentId", parentId);
+            }
+            resources.forEach(MDC::put);
+            if (attributes != null) {
+                EventResourcesImpl.forEach(attributes, MDC::put);
+            }
+            Logger logger = LoggerFactory.getLogger("stevlog");
+            switch (level) {
+            case ERROR:
+                if (throwable != null) {
+                    logger.error(event, throwable);
+                } else {
+                    logger.error(event);
+                }
+                break;
+            case WARN:
+                if (throwable != null) {
+                    logger.warn(event, throwable);
+                } else {
+                    logger.warn(event);
+                }
+                break;
+            case INFO:
+            default:
+                if (throwable != null) {
+                    logger.info(event, throwable);
+                } else {
+                    logger.info(event);
+                }
+                break;
+            }
+        } finally {
+            MDC.clear();
+        }
+    }
+
+    @Override
+    public void stash() {
+        throw new UnsupportedOperationException("TODO");
+    }
+
+    private List<Object> getAttributes() {
+        if (attributes == null) {
+            attributes = new ArrayList<>();
+        }
+        return attributes;
+    }
+
+    static String randomId() {
+        return Long.toString(
+                ThreadLocalRandom.current().nextLong(0x100000000000000L,
+                                                     0xFFFFFFFFFFFFFFFL),
+                16);
+    }
+
+    enum Level {
+        INFO,
+        WARN,
+        ERROR
+    }
+}

structured-event-log/src/main/java/org/apache/pulsar/structuredeventlog/slf4j/Slf4jStructuredEventLog.java

@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.structuredeventlog.slf4j;
+
+import org.apache.pulsar.structuredeventlog.Event;
+import org.apache.pulsar.structuredeventlog.EventResources;
+import org.apache.pulsar.structuredeventlog.EventResourcesImpl;
+import org.apache.pulsar.structuredeventlog.StructuredEventLog;
+
+public class Slf4jStructuredEventLog implements StructuredEventLog {
+    public static Slf4jStructuredEventLog INSTANCE = new Slf4jStructuredEventLog();
+
+    @Override
+    public Event newRootEvent() {
+        return new Slf4jEvent(null).traceId(Slf4jEvent.randomId());
+    }
+
+    @Override
+    public EventResources newEventResources() {
+        return new EventResourcesImpl(null);
+    }
+
+    @Override
+    public Event unstash() {
+        throw new UnsupportedOperationException("TODO");
+    }
+}