Logging

Author: slinkydeveloper

sdk-core/src/main/java/dev/restate/sdk/core/RequestProcessorImpl.java

@@ -221,7 +221,7 @@ private void onUserCodeResult(@Nullable Slice slice, @Nullable Throwable throwab
     try {
       if (throwable != null) {
         if (throwable instanceof TerminalException) {
-          LOG.info("Invocation completed with terminal error", throwable);
+          LOG.info("Invocation ended with terminal error", throwable);
           stateMachine.writeOutput((TerminalException) throwable);
           stateMachine.end();
         } else if (ExceptionUtils.containsAbortedExecutionException(throwable)) {
@@ -233,6 +233,7 @@ private void onUserCodeResult(@Nullable Slice slice, @Nullable Throwable throwab
       } else {
         stateMachine.writeOutput(Objects.requireNonNullElse(slice, Slice.EMPTY));
         stateMachine.end();
+        LOG.info("Invocation ended successfully");
       }
     } catch (Throwable e) {
       // Error happened when trying to write the final bits
@@ -249,6 +250,7 @@ private void onUserCodeResult(@Nullable Slice slice, @Nullable Throwable throwab
 
   private void startHandler() {
     state = State.RUNNING_HANDLER;
+    LOG.info("Invocation started");
 
     // Get vm input
     StateMachine.Input stateMachineInput = stateMachine.input();

sdk-core/src/main/java/dev/restate/sdk/core/legacy/State.java

@@ -202,8 +202,6 @@ default void hitSuspended(Collection<NotificationId> awaitingOn, StateContext st
   }
 
   default void end(StateContext stateContext) {
-    LOG.info("Invocation ended");
-
     stateContext.writeMessageOut(Protocol.EndMessage.getDefaultInstance());
     stateContext.getStateHolder().transition(new ClosedState());
 

sdk-core/src/main/java/dev/restate/sdk/core/legacy/WaitingStartState.java

@@ -46,9 +46,6 @@ public void onNewMessage(InvocationInput invocationInput, StateContext stateCont
                 : null));
     stateContext.setEagerState(new EagerState(startMessage));
 
-    // Tracing and logging setup
-    LOG.info("Start invocation");
-
     // Execute state transition
     stateContext.getStateHolder().transition(new WaitingReplayEntriesState());
   }

sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java

@@ -71,19 +71,9 @@ public final class FfmStateMachine implements StateMachine {
     // Load the native library BEFORE SharedCoreNative is class-initialized so its
     // SymbolLookup.loaderLookup() resolves the vm_* symbols.
     NativeLibraryLoader.ensureLoaded();
-    SharedCoreNative.init(defaultLogLevel());
-  }
-
-  private static int defaultLogLevel() {
-    // 0 trace, 1 debug, 2 info, 3 warn, 4 error
-    String level = System.getProperty("dev.restate.sharedcore.loglevel", "INFO");
-    return switch (level.trim().toUpperCase(java.util.Locale.ROOT)) {
-      case "TRACE" -> 0;
-      case "DEBUG" -> 1;
-      case "WARN" -> 3;
-      case "ERROR" -> 4;
-      default -> 2;
-    };
+    // Install the native tracing subscriber: filter at the host's configured level (so disabled
+    // callsites short-circuit in the core) and forward surviving events to Log4j2 via an upcall.
+    SharedCoreNative.init(NativeLogging.abiLevel(), NativeLogging.callbackStub());
   }
 
   public FfmStateMachine(HeadersAccessor headersAccessor) {

sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/NativeLogging.java

@@ -0,0 +1,161 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.sdk.core.statemachine.ffm;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.FunctionDescriptor;
+import java.lang.foreign.Linker;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.nio.charset.StandardCharsets;
+import java.util.Locale;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.spi.StandardLevel;
+
+/**
+ * Bridges the native {@code restate-sdk-shared-core} tracing output into Log4j2.
+ *
+ * <p>At startup the FFM state machine installs a {@code tracing} subscriber in the native core (see
+ * {@code init} in {@code rust/src/lib.rs}) parameterized with:
+ *
+ * <ul>
+ *   <li>a <b>max level</b> ({@link #abiLevel()}) derived from this class's Log4j2 logger, so the
+ *       core filters disabled events on its side — critically, its {@code #[instrument(level =
+ *       "trace")]} spans short-circuit before doing any work when trace is off;
+ *   <li>a <b>log callback</b> ({@link #callbackStub()}): an FFM upcall stub the core invokes for
+ *       every event that passes the filter.
+ * </ul>
+ *
+ * <p>The callback is invoked synchronously on the same thread that made the downcall into the core,
+ * so a native log line naturally inherits that thread's logging context (the {@code
+ * restateInvocationId} / {@code restateInvocationTarget} MDC set by the endpoint handler) — a
+ * Log4j2 pattern using {@code %X{restateInvocationId}} works unchanged.
+ *
+ * <p><b>Copy semantics.</b> The core hands the callback a borrowed {@code (ptr, len)} view over a
+ * buffer that is valid only for the duration of the call. We therefore eagerly copy the bytes into
+ * a JVM-heap {@code String} <b>during the call</b>: an async Log4j2 appender that retains the
+ * message then holds a self-contained Java object, never a view over native memory that has since
+ * been reused or freed.
+ */
+final class NativeLogging {
+
+  private NativeLogging() {}
+
+  /** Dedicated logger all native (shared-core) events are routed through. */
+  static final String LOGGER_NAME = "dev.restate.sdk.core.StateMachine";
+
+  private static final Logger LOG = LogManager.getLogger(LOGGER_NAME);
+
+  /**
+   * Optional override for the native max level, bypassing the Log4j2-derived value. Accepts {@code
+   * TRACE|DEBUG|INFO|WARN|ERROR}.
+   */
+  private static final String LEVEL_OVERRIDE_PROPERTY =
+      "dev.restate.sdk.core.statemachine.ffm.FfmStateMachine.loglevel";
+
+  // ABI ordinal -> Log4j2 level. Mirrors AbiLogLevel in rust/src/lib.rs.
+  private static final Level[] LEVELS = {
+    Level.TRACE, Level.DEBUG, Level.INFO, Level.WARN, Level.ERROR
+  };
+
+  private static final FunctionDescriptor CALLBACK_DESC =
+      FunctionDescriptor.ofVoid(ValueLayout.JAVA_INT, ValueLayout.ADDRESS, ValueLayout.JAVA_LONG);
+
+  /**
+   * Upcall entry point invoked by the native core for each tracing event. Must never throw across
+   * the FFI boundary, so everything is wrapped in a catch-all.
+   *
+   * @param level the {@link #LEVELS} ordinal
+   * @param msgPtr native pointer to UTF-8 bytes, valid only for this call
+   * @param msgLen length in bytes of the message at {@code msgPtr}
+   */
+  static void log(int level, MemorySegment msgPtr, long msgLen) {
+    try {
+      Level l = level >= 0 && level < LEVELS.length ? LEVELS[level] : Level.INFO;
+      // The native side already filtered by max level; this guards against per-logger config that
+      // is stricter than the max, and avoids the decode when the event would be dropped anyway.
+      if (!LOG.isEnabled(l)) {
+        return;
+      }
+      // Eagerly copy the native bytes into a JVM-heap String *now*, before returning. This is the
+      // single copy that makes the whole scheme safe: the native buffer is valid only for this
+      // call, so an async Log4j2 appender that retains the message must be handed a self-contained
+      // Java object, never a view over native memory.
+      byte[] bytes = msgPtr.reinterpret(msgLen).toArray(ValueLayout.JAVA_BYTE);
+      LOG.log(l, new String(bytes, StandardCharsets.UTF_8));
+    } catch (Throwable ignored) {
+      // Swallow: an exception unwinding into the native caller across `extern "C"` is UB.
+    }
+  }
+
+  /**
+   * Builds the upcall stub for {@link #log}. Uses {@link Arena#global()} (never freed): the native
+   * core keeps only the raw function pointer, which the GC cannot see, so the stub must outlive the
+   * process — an {@link Arena#ofAuto()} stub would be reclaimed as soon as the returned segment
+   * becomes unreachable (right after {@code init} returns), and the next event would call freed
+   * code.
+   */
+  static MemorySegment callbackStub() {
+    try {
+      MethodHandle handle =
+          MethodHandles.lookup()
+              .findStatic(
+                  NativeLogging.class,
+                  "log",
+                  MethodType.methodType(void.class, int.class, MemorySegment.class, long.class));
+      return Linker.nativeLinker().upcallStub(handle, CALLBACK_DESC, Arena.global());
+    } catch (NoSuchMethodException | IllegalAccessException e) {
+      throw new ExceptionInInitializerError(e);
+    }
+  }
+
+  /**
+   * The max native tracing level as an {@code AbiLogLevel} ordinal (0 trace .. 4 error), taken from
+   * the {@link #LEVEL_OVERRIDE_PROPERTY} system property when set, otherwise from the effective
+   * Log4j2 level of {@link #LOGGER_NAME}. Passed to native {@code init} so the core does not even
+   * produce events below it.
+   */
+  static int abiLevel() {
+    String override = System.getProperty(LEVEL_OVERRIDE_PROPERTY);
+    if (override != null) {
+      return switch (override.trim().toUpperCase(Locale.ROOT)) {
+        case "TRACE" -> 0;
+        case "DEBUG" -> 1;
+        case "WARN" -> 3;
+        case "ERROR" -> 4;
+        default -> 2;
+      };
+    }
+    return abiLevel(LOG.getLevel());
+  }
+
+  private static int abiLevel(Level level) {
+    if (level == null) {
+      return 2; // INFO
+    }
+    // Log4j2 intLevel decreases with severity (TRACE=600 .. ERROR=200); a logger set to level L
+    // permits L and everything more severe. Map that to the coarser native max level.
+    int i = level.intLevel();
+    if (i >= StandardLevel.TRACE.intLevel()) {
+      return 0;
+    } else if (i >= StandardLevel.DEBUG.intLevel()) {
+      return 1;
+    } else if (i >= StandardLevel.INFO.intLevel()) {
+      return 2;
+    } else if (i >= StandardLevel.WARN.intLevel()) {
+      return 3;
+    }
+    return 4;
+  }
+}

sdk-core/src/main/rust/src/lib.rs

@@ -59,28 +59,100 @@ use restate_sdk_shared_core::{
 };
 use std::borrow::Cow;
 use std::convert::Infallible;
+use std::ffi::c_void;
+use std::fmt::Write as _;
 use std::mem::MaybeUninit;
 use std::slice;
 use std::time::Duration;
+use tracing::field::{Field, Visit};
 use tracing::level_filters::LevelFilter;
-use tracing::Level;
+use tracing::{Event, Level, Subscriber};
+use tracing_subscriber::layer::{Context, Layer, SubscriberExt};
+use tracing_subscriber::registry::LookupSpan;
+use tracing_subscriber::util::SubscriberInitExt;
 
 // =========================================================================
 // Init & logging
 // =========================================================================
 
+/// Zero-copy log sink installed by the host: `(level, msg_ptr, msg_len)`. `level` is the
+/// [`AbiLogLevel`] ordinal; the `(ptr, len)` pair is a UTF-8 view into a Rust-owned, thread-local
+/// buffer that is valid **only for the synchronous duration of this call**. The callee must copy
+/// the bytes out before returning and must never retain the pointer — no allocation crosses the
+/// boundary and nothing is freed here (contrast the owned `Slice` results). It must also never
+/// unwind (the Java upcall swallows all exceptions), as unwinding across `extern "C"` is UB.
+type LogCallback = extern "C" fn(level: u32, msg_ptr: *const u8, msg_len: usize);
+
+/// Install the process-global tracing subscriber and panic hook. `level` is the max [`AbiLogLevel`]
+/// to emit — set from the host's logging configuration so disabled callsites (e.g. the core's
+/// `#[instrument(level = "trace")]` spans) short-circuit before doing any work. Events that pass the
+/// filter are formatted into a thread-local buffer and forwarded to the host via [`LogCallback`]
+/// (zero-copy).
+///
+/// This must be called exactly once. `log_callback` must be non-null (the host always installs a
+/// sink), and a second call — or any pre-existing global subscriber — **panics**: silently ignoring
+/// it would hide a real wiring bug and leave logs going nowhere.
 #[no_mangle]
-pub unsafe extern "C" fn init(level: u32) {
+pub unsafe extern "C" fn init(level: u32, log_callback: *const c_void) {
     std::panic::set_hook(Box::new(|panic| {
         eprintln!("[restate-shared-core] core panicked: {panic}");
     }));
+    assert!(
+        !log_callback.is_null(),
+        "init called with a null log callback"
+    );
     let level: Level = AbiLogLevel::from(level).into();
-    let _ = tracing_subscriber::fmt()
-        .with_ansi(false)
-        .with_writer(std::io::stderr)
-        .with_max_level(LevelFilter::from_level(level))
-        .with_target(level == Level::TRACE)
-        .try_init();
+    let max = LevelFilter::from_level(level);
+
+    // SAFETY: the host passed a non-null upcall stub with the `LogCallback` signature. On all
+    // supported targets a data pointer and a C function pointer share a representation.
+    let callback: LogCallback = std::mem::transmute(log_callback);
+    // `.init()` (not `try_init`) panics if a global subscriber is already set — see the doc above.
+    tracing_subscriber::registry()
+        .with(JavaLogLayer { callback }.with_filter(max))
+        .init();
+}
+
+/// A `tracing` layer that forwards each event to the host [`LogCallback`]. The event's message and
+/// fields are rendered into a `String` and handed to the callback as a borrowed `(ptr, len)` view;
+/// the host copies the bytes out synchronously (which is what makes it safe against async log
+/// appenders), and the `String` is dropped as soon as the callback returns. Events are already
+/// filtered by max level, so allocating a fresh buffer per event is not on any hot path.
+struct JavaLogLayer {
+    callback: LogCallback,
+}
+
+impl<S> Layer<S> for JavaLogLayer
+where
+    S: Subscriber + for<'a> LookupSpan<'a>,
+{
+    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
+        let level = AbiLogLevel::from(*event.metadata().level()) as u32;
+        let mut msg = String::new();
+        event.record(&mut MessageVisitor { buf: &mut msg });
+        (self.callback)(level, msg.as_ptr(), msg.len());
+    }
+}
+
+/// Renders a `tracing` event into `buf` as `message key=value ...` (the message field verbatim,
+/// other fields as `key=value`). Span context is intentionally omitted — at INFO/DEBUG the core's
+/// trace-level spans don't exist, and per-invocation context (invocation id, target) already rides
+/// in the host's logging MDC via the same-thread upcall.
+struct MessageVisitor<'a> {
+    buf: &'a mut String,
+}
+
+impl Visit for MessageVisitor<'_> {
+    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+        if !self.buf.is_empty() {
+            self.buf.push(' ');
+        }
+        if field.name() == "message" {
+            let _ = write!(self.buf, "{value:?}");
+        } else {
+            let _ = write!(self.buf, "{}={:?}", field.name(), value);
+        }
+    }
 }
 
 #[repr(u32)]
@@ -104,6 +176,18 @@ impl From<u32> for AbiLogLevel {
     }
 }
 
+impl From<Level> for AbiLogLevel {
+    fn from(value: Level) -> Self {
+        match value {
+            Level::TRACE => AbiLogLevel::Trace,
+            Level::DEBUG => AbiLogLevel::Debug,
+            Level::INFO => AbiLogLevel::Info,
+            Level::WARN => AbiLogLevel::Warn,
+            Level::ERROR => AbiLogLevel::Error,
+        }
+    }
+}
+
 impl From<AbiLogLevel> for Level {
     fn from(value: AbiLogLevel) -> Self {
         match value {