participant-integration-api: Fix completion debug log output. (#10415)

* participant-integration-api: Fix completion debug log output.

So it doesn't embed Scala class names.

**Before:**

```
DEBUG c.d.p.a.s.ApiCommandCompletionService - Responding with completions: List(LoggingEntries(Map(commandId -> OfString(SemanticDoubleSpendBasic-alpha-2992c8731c2f-command-1), statusCode -> OfString(0)))), context: {parties: ["SemanticDoubleSpendBasic-alpha-2992c8731c2f-party-1"], offset: "000000000000000a0000000000000000"}
```

**After:**

```
DEBUG c.d.p.a.s.ApiCommandCompletionService - Responding with completions: [{commandId: SemanticDoubleSpendBasic-alpha-29a7c9d39f7b-command-1, statusCode: 0}], context: {parties: ["SemanticDoubleSpendBasic-alpha-29a7c9d39f7b-party-1"], offset: "000000000000000a0000000000000000"}
```

CHANGELOG_BEGIN
CHANGELOG_END

* participant-integration-api: Move completion stream logs into context.

It's structured data; let's treat it as such.

ledger/participant-integration-api/src/main/scala/platform/apiserver/services/ApiCommandCompletionService.scala

@@ -16,10 +16,11 @@ import com.daml.ledger.api.v1.command_completion_service._
 import com.daml.ledger.api.validation.PartyNameChecker
 import com.daml.ledger.participant.state.index.v2.IndexCompletionsService
 import com.daml.logging.LoggingContext.withEnrichedLoggingContext
-import com.daml.logging.entries.LoggingEntries
+import com.daml.logging.entries.{LoggingEntries, LoggingValue}
 import com.daml.logging.{ContextualizedLogger, LoggingContext}
 import com.daml.metrics.Metrics
 import com.daml.platform.api.grpc.GrpcApiService
+import com.daml.platform.apiserver.services.ApiCommandCompletionService._
 import com.daml.platform.server.api.services.domain.CommandCompletionService
 import com.daml.platform.server.api.services.grpc.GrpcCommandCompletionService
 import io.grpc.ServerServiceDefinition
@@ -36,6 +37,8 @@ private[apiserver] final class ApiCommandCompletionService private (
     loggingContext: LoggingContext,
 ) extends CommandCompletionService {
 
+  import Logging._
+
   private val logger = ContextualizedLogger.get(this.getClass)
 
   private val subscriptionIdCounter = new AtomicLong()
@@ -52,27 +55,18 @@ private[apiserver] final class ApiCommandCompletionService private (
 
         completionsService
           .getCompletions(offset, request.applicationId, request.parties)
-          .via(logger.debugStream(completionsLoggable))
+          .via(
+            logger.enrichedDebugStream(
+              "Responding with completions.",
+              response => LoggingEntries("response" -> responseToLoggingValue(response)),
+            )
+          )
           .via(logger.logErrorsOnStream)
           .via(StreamMetrics.countElements(metrics.daml.lapi.streams.completions))
     }
 
-  private def completionsLoggable(response: CompletionStreamResponse): String =
-    s"Responding with completions: ${response.completions.toList
-      .map(c => singleCompletionLoggable(c.commandId, c.status.map(_.code)))}"
-
-  private def singleCompletionLoggable(
-      commandId: String,
-      statusCode: Option[Int],
-  ): LoggingEntries =
-    LoggingEntries(
-      logging.commandId(commandId),
-      "statusCode" -> statusCode.fold("")(_.toString),
-    )
-
   override def getLedgerEnd(ledgerId: domain.LedgerId): Future[LedgerOffset.Absolute] =
     completionsService.currentLedgerEnd().andThen(logger.logErrorsOnCall[LedgerOffset.Absolute])
-
 }
 
 private[apiserver] object ApiCommandCompletionService {
@@ -93,4 +87,16 @@ private[apiserver] object ApiCommandCompletionService {
         CommandCompletionServiceGrpc.bindService(this, executionContext)
     }
   }
+
+  private object Logging {
+    def responseToLoggingValue(response: CompletionStreamResponse): LoggingValue =
+      LoggingValue.OfIterable(
+        response.completions.view.map(completion =>
+          LoggingValue.Nested.fromEntries(
+            "commandId" -> completion.commandId,
+            "statusCode" -> completion.status.map(_.code),
+          )
+        )
+      )
+  }
 }

libs-scala/contextualized-logging/src/main/scala/com/digitalasset/logging/ContextualizedLogger.scala

@@ -6,6 +6,7 @@ package com.daml.logging
 import akka.NotUsed
 import akka.stream.scaladsl.Flow
 import com.daml.grpc.GrpcException
+import com.daml.logging.entries.LoggingEntries
 import io.grpc.Status
 import org.slf4j.{Logger, LoggerFactory}
 
@@ -82,4 +83,15 @@ final class ContextualizedLogger private (val withoutContext: Logger) {
       item
     }
 
+  def enrichedDebugStream[Out](
+      msg: String,
+      withContext: Out => LoggingEntries,
+  )(implicit loggingContext: LoggingContext): Flow[Out, Out, NotUsed] =
+    Flow[Out].map { item =>
+      LoggingContext.withEnrichedLoggingContextFrom(withContext(item)) { implicit loggingContext =>
+        debug(msg)
+        item
+      }
+    }
+
 }

libs-scala/logging-entries/src/main/scala/com/daml/logging/entries/LoggingValue.scala

@@ -32,6 +32,10 @@ object LoggingValue {
 
   final case class Nested(entries: LoggingEntries) extends LoggingValue
 
+  object Nested {
+    def fromEntries(entries: LoggingEntry*): Nested = Nested(LoggingEntries(entries: _*))
+  }
+
   final case class OfJson(json: JsValue) extends LoggingValue
 
   @inline