Fix persisted headers, quiet noisy logs (#124)

Co-authored-by: Antonio Yuen <antonio.yuen@namely.com>

Author: Zen Yui

code/service/src/main/resources/logback.xml

@@ -38,6 +38,8 @@
     <logger name="org.apache.kafka" level="ERROR"/>
     <!-- Turn down Slick noise-->
     <logger name="scala.slick" level="ERROR"/>
+    <!-- Turn down kamon -->
+    <logger name="kamon.jaeger" level="WARN"/>
 
     <root level="${LOG_LEVEL}">
         <appender-ref ref="ASYNCJSON"/>

code/service/src/main/scala/com/namely/chiefofstate/GrpcServiceImpl.scala

@@ -49,11 +49,9 @@ class GrpcServiceImpl(sys: ActorSystem,
       )
     } else {
 
-      val meta: Try[Map[String, Any]] = pluginManager.run(in, MetadataUtil.makeMeta(metadata))
-
-      meta match {
-        case Success(m) =>
-          // get the headers to forward
+      Future
+        .fromTry(pluginManager.run(in, MetadataUtil.makeMeta(metadata)))
+        .flatMap(meta => {
           val propagatedHeaders: Seq[RemoteCommand.Header] = metadata.asList
             // filter to relevant headers
             .filter({ case (k, _) => sendCommandSettings.propagatedHeaders.contains(k) })
@@ -75,15 +73,14 @@ class GrpcServiceImpl(sys: ActorSystem,
             .withCommand(in.getCommand)
             .withHeaders(propagatedHeaders)
 
-          sendCommand(in.entityId, remoteCommand, m)
+          sendCommand(in.entityId, remoteCommand, meta)
             .map((stateWrapper: StateWrapper) => {
               ProcessCommandResponse(
                 state = stateWrapper.state,
                 meta = stateWrapper.meta.map(Util.toCosMetaData)
               )
             })
-        case Failure(e) => Future.fromTry(Failure(e))
-      }
+        })
     }
   }
 

code/service/src/main/scala/com/namely/chiefofstate/plugin/PersistHeaders.scala

@@ -5,33 +5,59 @@ import com.namely.protobuf.chiefofstate.plugins.persistedheaders.v1.headers.{Hea
 import com.namely.protobuf.chiefofstate.v1.service.ProcessCommandRequest
 import io.grpc.Metadata
 import scala.jdk.CollectionConverters._
+import scala.collection.mutable
+import org.slf4j.{Logger, LoggerFactory}
 
 private[this] class PersistHeaders(persistedHeaders: Seq[String]) extends PluginBase {
 
+  import PersistHeaders.logger
+
   override val pluginId: String = "persisted_headers.v1"
 
   override def run(processCommandRequest: ProcessCommandRequest, metadata: Metadata): Option[com.google.protobuf.any.Any] = {
-    val headers: Seq[Header] = persistedHeaders.flatMap(header => {
-      if(header.endsWith("-bin")) {
-        val bytesKey: Metadata.Key[Array[Byte]] = Metadata.Key.of(header, Metadata.BINARY_BYTE_MARSHALLER)
-        metadata.getAll[Array[Byte]](bytesKey).asScala
-          .map(b => Header().withKey(header).withBytesValue(ByteString.copyFrom(b)))
+
+    val capturedHeaders: mutable.ListBuffer[Header] = mutable.ListBuffer.empty[Header]
+
+    persistedHeaders.foreach(key => {
+      if(key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+        val bytesKey: Metadata.Key[Array[Byte]] = Metadata.Key.of(key, Metadata.BINARY_BYTE_MARSHALLER)
+        val byteValues = metadata.getAll[Array[Byte]](bytesKey)
+
+        if(byteValues != null) {
+          byteValues.forEach(byteArray => {
+            val byteString = ByteString.copyFrom(byteArray)
+            logger.debug(s"persisting header=${key}, type=bytes")
+            capturedHeaders.addOne(Header(key=key).withBytesValue(byteString))
+          })
+        }
       } else {
-        val stringKey: Metadata.Key[String] = Metadata.Key.of(header, Metadata.ASCII_STRING_MARSHALLER)
-        metadata.getAll[String](stringKey).asScala
-          .map(s => Header().withKey(header).withStringValue(s))
+        val stringKey: Metadata.Key[String] = Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER)
+        val stringValues = metadata.getAll[String](stringKey)
+
+        if(stringValues != null) {
+          stringValues.forEach(stringValue => {
+            logger.debug(s"persisting header=$key, type=string, value=$stringValue")
+            capturedHeaders.addOne(Header(key=key).withStringValue(stringValue))
+          })
+        }
       }
     })
 
-    Some(com.google.protobuf.any.Any.pack(Headers().withHeaders(headers)))
+    if(capturedHeaders.nonEmpty) {
+      Some(com.google.protobuf.any.Any.pack(Headers().withHeaders(capturedHeaders.toSeq)))
+    } else {
+      None
+    }
   }
 }
 
 object PersistHeaders extends PluginFactory {
 
   val envName: String = "COS_WRITE_PERSISTED_HEADERS"
 
-  lazy val persistedHeaders: Seq[String] = sys.env.get(envName)
+  val logger: Logger = LoggerFactory.getLogger(getClass)
+
+  def persistedHeaders: Seq[String] = sys.env.get(envName)
     .map(_.split(",").map(_.trim).toSeq)
     .getOrElse(Seq.empty[String])
 

code/service/src/main/scala/com/namely/chiefofstate/plugin/PluginManager.scala

@@ -5,7 +5,7 @@ import com.google.protobuf.any.Any
 import com.namely.protobuf.chiefofstate.v1.service.ProcessCommandRequest
 import com.typesafe.config.Config
 import io.grpc.{Metadata, Status}
-
+import org.slf4j.{Logger, LoggerFactory}
 import scala.reflect.runtime.universe
 import scala.util.{Failure, Success, Try}
 
@@ -15,6 +15,8 @@ import scala.util.{Failure, Success, Try}
  * @param plugins Sequence of PluginBase
  */
 case class PluginManager(plugins: Seq[PluginBase]) {
+  import PluginManager.logger
+
   /**
    * Given a PluginManager instance, a ProcessCommandRequest instance and io.grpc.Metadata instance, folds left
    * throughout all the plugins and runs them one at a time. The results are mapped into a Map of pluginId -> protoAny.
@@ -34,7 +36,15 @@ case class PluginManager(plugins: Seq[PluginBase]) {
       }
       pluginRun match {
         case Success(m) => Try(metaMap.get ++ m)
-        case Failure(e) => Failure(new GrpcServiceException(status = Status.ABORTED.withDescription(e.getMessage)))
+        case Failure(e: GrpcServiceException) =>
+          logger.error(s"plugin '${plugin.pluginId}' failed with ${e.getClass.getName}: ${e.getStatus.toString}")
+          Failure(e)
+        case Failure(e: Throwable) =>
+          val errMsg = s"plugin ${plugin.pluginId} failed due to ${e.getClass.getName}: ${e.getMessage}"
+          logger.error(errMsg)
+          val status = Status.INTERNAL.withDescription(e.getMessage)
+          val err = new GrpcServiceException(status=status)
+          Failure(err)
       }
     })
   }
@@ -47,6 +57,8 @@ object PluginManager {
 
   final val HOCON_PATH: String = "chief-of-state.plugin-settings.enable-plugins"
 
+  lazy val logger: Logger = LoggerFactory.getLogger(getClass)
+
   /**
    * Default COS Plugins
    */

…fstate/plugin/PersistedHeadersSpec.scala …fofstate/plugin/PersistHeadersSpec.scalacode/service/src/test/scala/com/namely/chiefofstate/plugin/PersistedHeadersSpec.scala renamed to code/service/src/test/scala/com/namely/chiefofstate/plugin/PersistHeadersSpec.scala code/service/src/test/scala/com/namely/chiefofstate/plugin/PersistedHeadersSpec.scala renamed to code/service/src/test/scala/com/namely/chiefofstate/plugin/PersistHeadersSpec.scala

@@ -8,7 +8,12 @@ import com.namely.protobuf.chiefofstate.plugins.persistedheaders.v1.headers.Head
 import com.namely.protobuf.chiefofstate.plugins.persistedheaders.v1.headers.{Header, Headers}
 import com.namely.protobuf.chiefofstate.v1.service.ProcessCommandRequest
 
-class PersistedHeadersSpec extends TestSpec with MockFactory {
+class PersistHeadersSpec extends TestSpec with MockFactory {
+
+  override def beforeEach(): Unit = {
+    super.beforeEach()
+    EnvironmentHelper.clearEnv()
+  }
 
   "Persistedheaders" should {
     val fooKeyName: String = "foo"
@@ -40,7 +45,7 @@ class PersistedHeadersSpec extends TestSpec with MockFactory {
     metadata.put(bazKey, baz)
 
     "return the a string and byte header" in {
-      EnvironmentHelper.setEnv(PersistHeaders.envName, "foo,bar-bin")
+      EnvironmentHelper.setEnv(PersistHeaders.envName, "foo,bar-bin,not-a-key")
 
       val actual: Headers = PersistHeaders
         .apply()
@@ -52,5 +57,15 @@ class PersistedHeadersSpec extends TestSpec with MockFactory {
 
       actual should be (expected)
     }
+
+    "return an empty header" in {
+      EnvironmentHelper.setEnv(PersistHeaders.envName, "not-a-key,not-a-key-bin")
+
+      val actual: Option[com.google.protobuf.any.Any] = PersistHeaders
+        .apply()
+        .run(processCommandRequest, metadata)
+
+      actual should be (None)
+    }
   }
 }

code/service/src/test/scala/com/namely/chiefofstate/plugin/PluginManagerSpec.scala

@@ -11,6 +11,7 @@ import org.scalatest.Assertion
 import org.scalatest.matchers.should.Matchers
 
 import scala.util.Try
+import io.grpc.Status
 
 private[this] class MockPlugin1() extends PluginBase {
   override val pluginId: String = "MockPluginBase"
@@ -124,10 +125,22 @@ class PluginManagerSpec extends TestSpec {
 
       "return a failure" in {
         val mockPluginBase: PluginBase = mock[PluginBase]
-        (mockPluginBase.run _).expects(processCommandRequest, metadata).throws(new RuntimeException("test"))
+
+        (mockPluginBase.run _)
+          .expects(processCommandRequest, metadata)
+          .throws(new RuntimeException("test"))
+
+        (() => mockPluginBase.pluginId)
+          .expects()
+          .returning("some-plugin-id")
 
         val pluginManager: PluginManager = new PluginManager(Seq(mockPluginBase))
-        intercept[GrpcServiceException](pluginManager.run(processCommandRequest, metadata).get)
+
+        val actual = pluginManager.run(processCommandRequest, metadata)
+        val actualError = intercept[GrpcServiceException](actual.get)
+
+        actualError.getStatus.getCode shouldBe(Status.Code.INTERNAL)
+        actualError.getStatus.getDescription() shouldBe("test")
       }
     }
   }