path: root/kamon-core/src/main/scala/kamon/context/HttpPropagation.scala

package kamon
package context

import com.typesafe.config.Config
import org.slf4j.LoggerFactory

import scala.reflect.ClassTag
import scala.util.control.NonFatal
import scala.util.{Failure, Success}

/**
  * Context Propagation for HTTP transports. When using HTTP transports all the context related information is
  * read from and written to HTTP headers. The context information may be included in the following directions:
  *   - Incoming: Used for HTTP requests coming into this service. Implicitly used when using HttpPropagation.read.
  *   - Outgoing: Used for HTTP requests leaving this service.
  *   - Returning: Used for HTTP responses send back to clients of this service.
  */
trait HttpPropagation {

  /**
    * Uses the provided [[HttpPropagation.HeaderReader]] to read as many HTTP Headers as necessary and create a
    * [[Context]] instance. The way in which context tags and entries are read from and written to HTTP Headers is
    * implementation specific.
    *
    * @param reader Wrapper on the HTTP message from which headers are read.
    * @return The decoded Context instance. If no entries or tags could be read from the HTTP message then an
    *          empty context is returned instead.
    */
  def readContext(reader: HttpPropagation.HeaderReader): Context

  /**
    * Writes the tags and entries from the supplied context using the supplied [[HttpPropagation.HeaderWriter]]
    * instance. The way in which context tags and entries are read from and written to HTTP Headers is implementation
    * specific.
    *
    * Implementations are expected to produce side effects on the wrapped HTTP Messages.
    *
    * @param context Context instance to be written.
    * @param writer Wrapper on the HTTP message that will carry the context headers.
    * @param direction Write direction. It can be either Outgoing or Returning.
    */
  def writeContext(context: Context, writer: HttpPropagation.HeaderWriter, direction: HttpPropagation.Direction.Write): Unit

}

object HttpPropagation {

  /**
    * Encapsulates logic required to read a single context entry from HTTP headers. Implementations of this trait
    * must be aware of the entry they are able to read and the HTTP headers required to do so.
    */
  trait EntryReader {

    /**
      * Tries to read a context entry from HTTP headers. If a context entry is successfully read, implementations
      * must return an updated context instance that includes such entry. If no entry could be read simply return
      * context instance that was passed in, untouched.
      *
      * @param reader Wrapper on the HTTP message from which headers are read.
      * @param context Current context.
      * @return Either the original context passed in or a modified version of it, including the read entry.
      */
    def readEntry(reader: HttpPropagation.HeaderReader, context: Context): Context
  }

  /**
    * Encapsulates logic required to write a single context entry to HTTP headers. Implementations of this trait
    * must be aware of the entry they are able to write and the HTTP headers required to do so.
    */
  trait EntryWriter {

    /**
      * Tries to write a context entry into HTTP headers.
      *
      * @param context The context from which entries should be written.
      * @param writer Wrapper on the HTTP message that will carry the context headers.
      * @param direction Write direction. It can be either Outgoing or Returning.
      */
    def writeEntry(context: Context, writer: HttpPropagation.HeaderWriter, direction: Direction.Write): Unit
  }


  /**
    * Wrapper that reads HTTP headers from HTTP a message.
    */
  trait HeaderReader {

    /**
      * Reads an HTTP header value
      *
      * @param header HTTP header name
      * @return The HTTP header value, if present.
      */
    def readHeader(header: String): Option[String]
  }

  /**
    * Wrapper that writes HTTP headers to a HTTP message.
    */
  trait HeaderWriter {

    /**
      * Writes a HTTP header into a HTTP message.
      *
      * @param header HTTP header name.
      * @param value HTTP header value.
      */
    def writeHeader(header: String, value: String): Unit
  }


  /**
    * Create a new default HttpPropagation instance from the provided configuration.
    *
    * @param config HTTP propagation channel configuration
    * @return A newly constructed HttpPropagation instance.
    */
  def from(config: Config, classLoading: ClassLoading): HttpPropagation = {
    new HttpPropagation.Default(Components.from(config, classLoading))
  }

  /**
    * Default HTTP Propagation in Kamon.
    */
  final class Default(components: Components) extends HttpPropagation {
    private val log = LoggerFactory.getLogger(classOf[HttpPropagation.Default])

    /**
      * Reads context tags and entries on the following order:
      *   - Read all context tags from the context tags header.
      *   - Read all context tags with explicit mappings. This overrides any tag from the previous step in case
      *     of a tag key clash.
      *   - Read all context entries using the incoming entries configuration.
      */
    override def readContext(reader: HeaderReader): Context = {
      val tags = Map.newBuilder[String, String]

      // Tags encoded together in the context tags header.
      try {
        reader.readHeader(components.tagsHeaderName).foreach { contextTagsHeader =>
          contextTagsHeader.split(";").foreach(tagData => {
            val tagPair = tagData.split("=")
            if (tagPair.length == 2) {
              tags += (tagPair(0) -> tagPair(1))
            }
          })
        }
      } catch {
        case NonFatal(t) => log.warn("Failed to read the context tags header", t.asInstanceOf[Any])
      }

      // Tags explicitly mapped on the tags.mappings configuration.
      components.tagsMappings.foreach {
        case (tagName, httpHeader) =>
          try {
            reader.readHeader(httpHeader).foreach(tagValue => tags += (tagName -> tagValue))
          } catch {
            case NonFatal(t) => log.warn("Failed to read mapped tag [{}]", tagName, t.asInstanceOf[Any])
          }
      }

      // Incoming Entries
      components.incomingEntries.foldLeft(Context.of(tags.result())) {
        case (context, (entryName, entryDecoder)) =>
          var result = context
          try {
            result = entryDecoder.readEntry(reader, context)
          } catch {
            case NonFatal(t) => log.warn("Failed to read entry [{}]", entryName.asInstanceOf[Any], t.asInstanceOf[Any])
          }

          result
      }
    }

    /**
      * Writes context tags and entries
      */
    override def writeContext(context: Context, writer: HeaderWriter, direction: Direction.Write): Unit = {
      val keys = direction match {
        case Direction.Outgoing => components.outgoingEntries
        case Direction.Returning => components.returningEntries
      }

      val contextTagsHeader = new StringBuilder()
      def appendTag(key: String, value: String): Unit = {
        contextTagsHeader
          .append(key)
          .append('=')
          .append(value)
          .append(';')
      }

      // Write tags with specific mappings or append them to the context tags header.
      context.tags.foreach {
        case (tagKey, tagValue) => components.tagsMappings.get(tagKey) match {
          case Some(mappedHeader) => writer.writeHeader(mappedHeader, tagValue)
          case None => appendTag(tagKey, tagValue)
        }
      }

      // Write the context tags header.
      if(contextTagsHeader.nonEmpty) {
        writer.writeHeader(components.tagsHeaderName, contextTagsHeader.result())
      }

      // Write entries for the specified direction.
      keys.foreach {
        case (entryName, entryWriter) =>
          try {
            entryWriter.writeEntry(context, writer, direction)
          } catch {
            case NonFatal(t) => log.warn("Failed to write entry [{}] due to: {}", entryName.asInstanceOf[Any], t.asInstanceOf[Any])
          }
      }
    }
  }

  /**
    * Propagation direction. Used to decide whether incoming, outgoing or returning keys must be used to
    * propagate context.
    */
  sealed trait Direction
  object Direction {

    /**
      * Marker trait for all directions that require write operations.
      */
    sealed trait Write

    /**
      * Requests coming into this service.
      */
    case object Incoming extends Direction

    /**
      * Requests going from this service to others.
      */
    case object Outgoing extends Direction with Write

    /**
      * Responses sent from this service to clients.
      */
    case object Returning extends Direction with Write
  }


  case class Components(
    tagsHeaderName: String,
    tagsMappings: Map[String, String],
    incomingEntries: Map[String, HttpPropagation.EntryReader],
    outgoingEntries: Map[String, HttpPropagation.EntryWriter],
    returningEntries: Map[String, HttpPropagation.EntryWriter]
  )

  object Components {
    private val log = LoggerFactory.getLogger(classOf[HttpPropagation.Components])

    def from(config: Config, classLoading: ClassLoading): Components = {
      def buildInstances[ExpectedType : ClassTag](mappings: Map[String, String]): Map[String, ExpectedType] = {
        val entryReaders = Map.newBuilder[String, ExpectedType]

        mappings.foreach {
          case (contextKey, readerClass) => classLoading.createInstance[ExpectedType](readerClass, Nil) match {
            case Success(readerInstance) => entryReaders += (contextKey -> readerInstance)
            case Failure(exception) => log.warn("Failed to instantiate {} [{}] due to []",
              implicitly[ClassTag[ExpectedType]].runtimeClass.getName, readerClass, exception)
          }
        }

        entryReaders.result()
      }

      val tagsHeaderName = config.getString("tags.header-name")
      val tagsMappings = config.getConfig("tags.mappings").pairs
      val incomingEntries = buildInstances[HttpPropagation.EntryReader](config.getConfig("entries.incoming").pairs)
      val outgoingEntries = buildInstances[HttpPropagation.EntryWriter](config.getConfig("entries.outgoing").pairs)
      val returningEntries = buildInstances[HttpPropagation.EntryWriter](config.getConfig("entries.returning").pairs)

      Components(tagsHeaderName, tagsMappings, incomingEntries, outgoingEntries, returningEntries)
    }
  }

}