Merge pull request #81 from ExpediaDotCom/spanSizeValidation

Updating spanSize validation logic to skip certain tags and services

Author: absrivastava

commons/src/main/scala/com/expedia/www/haystack/collector/commons/ProtoSpanExtractor.scala

@@ -24,14 +24,14 @@ import java.util.concurrent.ConcurrentHashMap
 
 import com.expedia.open.tracing.{Span, Tag}
 import com.expedia.www.haystack.collector.commons.ProtoSpanExtractor._
-import com.expedia.www.haystack.collector.commons.config.{ExtractorConfiguration, Format, SpanMaxSize}
+import com.expedia.www.haystack.collector.commons.config.{ExtractorConfiguration, Format}
 import com.expedia.www.haystack.collector.commons.record.{KeyValueExtractor, KeyValuePair}
 import com.expedia.www.haystack.span.decorators.SpanDecorator
 import com.google.protobuf.util.JsonFormat
 import org.slf4j.Logger
 
-import scala.util.{Failure, Success, Try}
 import scala.collection.JavaConverters._
+import scala.util.{Failure, Success, Try}
 
 object ProtoSpanExtractor {
   private val DaysInYear1970 = 365
@@ -45,10 +45,9 @@ object ProtoSpanExtractor {
   val TraceIdIsRequired = "Trace ID is required: serviceName=[%s] operationName=[%s]"
   val StartTimeIsInvalid = "Start time [%d] is invalid: serviceName=[%s] operationName=[%s]"
   val DurationIsInvalid = "Duration [%d] is invalid: serviceName=[%s] operationName=[%s]"
-  val SpanSizeLimitExceeded = "Span Size Limit Exceeded: serviceName=[%s] operationName=[%s] traceId=[%s] spanSize=[%d]"
+  val SpanSizeLimitExceeded = "Span Size Limit Exceeded: serviceName=[%s] operationName=[%s] traceId=[%s] spanSize=[%d] probableTags=[%s]"
 
   val ServiceNameVsTtlAndOperationNames = new ConcurrentHashMap[String, TtlAndOperationNames]
-  val MaximumOperationNameCount = 1000
   val OperationNameCountExceededMeterName = "operation.name.count.exceeded"
 }
 
@@ -60,7 +59,7 @@ class ProtoSpanExtractor(extractorConfiguration: ExtractorConfiguration,
 
   private val invalidSpanMeter = metricRegistry.meter("invalid.span")
   private val validSpanMeter = metricRegistry.meter("valid.span")
-  private val spanSizeLimitExceeded = metricRegistry.meter("sizeLimitExceeded.span")
+  private val spanSizeLimitExceededMeter = metricRegistry.meter("sizeLimitExceeded.span")
 
   override def configure(): Unit = ()
 
@@ -89,12 +88,12 @@ class ProtoSpanExtractor(extractorConfiguration: ExtractorConfiguration,
   }
 
   def validateSpanSize(span: Span): Try[Span] = {
-    if (extractorConfiguration.spanValidation.spanMaxSize.enable)
-      {
-        val spanSize = span.toByteArray.length
-        val maxSizeLimit = extractorConfiguration.spanValidation.spanMaxSize.maxSizeLimit
-        validate(span, spanSize, SpanSizeLimitExceeded, maxSizeLimit)
-      }
+    if (extractorConfiguration.spanValidation.spanMaxSize.enable
+      && !extractorConfiguration.spanValidation.spanMaxSize.skipServices.contains(span.getServiceName.toLowerCase)) {
+      val spanSize = span.toByteArray.length
+      val maxSizeLimit = extractorConfiguration.spanValidation.spanMaxSize.maxSizeLimit
+      validate(span, spanSize, SpanSizeLimitExceeded, maxSizeLimit)
+    }
     else
       Success(span)
   }
@@ -141,24 +140,39 @@ class ProtoSpanExtractor(extractorConfiguration: ExtractorConfiguration,
                        highestValidValue: Int): Try[Span] = {
 
     if (valueToValidate > highestValidValue) {
-      spanSizeLimitExceeded.mark()
-      LOGGER.debug(msg.format(span.getServiceName, span.getOperationName, span.getTraceId, valueToValidate))
-      Success(truncateTags(span))
+      spanSizeLimitExceededMeter.mark()
+      LOGGER.debug(msg.format(span.getServiceName, span.getOperationName, span.getTraceId, valueToValidate, getProbableTagsExceedingSizeLimit(span)))
+      if (extractorConfiguration.spanValidation.spanMaxSize.logOnly) {
+        Success(span)
+      } else {
+        Success(truncateTags(span))
+      }
     }
     else {
       Success(span)
     }
   }
 
-  private def truncateTags(span : Span): Span = {
-    val errorTag = span.getTagsList.asScala.filter(tag => tag.getKey.equalsIgnoreCase("error"))
-    val spanBuilder = span.toBuilder
-    val messsageTagKey = extractorConfiguration.spanValidation.spanMaxSize.infoTagKey
-    val messageTagValue = extractorConfiguration.spanValidation.spanMaxSize.infoTagValue
+  private def getProbableTagsExceedingSizeLimit(span: Span): String = {
+    span.getTagsList.asScala
+      .filter(tag => tag.getVStrBytes.size > extractorConfiguration.spanValidation.spanMaxSize.maxSizeLimit)
+      .map(_.getKey)
+      .mkString(", ")
+  }
+
+  private def truncateTags(span: Span): Span = {
+    val skippedTags = span.getTagsList.asScala
+      .filter(tag => extractorConfiguration.spanValidation.spanMaxSize.skipTags.contains(tag.getKey.toLowerCase))
 
+    val spanBuilder = span.toBuilder
     spanBuilder.clearTags()
-    errorTag.foreach(tag => spanBuilder.addTags(tag))
-    spanBuilder.addTags(Tag.newBuilder().setKey(messsageTagKey).setVStr(messageTagValue))
+
+    skippedTags.foreach(spanBuilder.addTags)
+
+    val truncateTagKey = extractorConfiguration.spanValidation.spanMaxSize.infoTagKey
+    val truncateTagValue = extractorConfiguration.spanValidation.spanMaxSize.infoTagValue
+    spanBuilder.addTags(Tag.newBuilder().setKey(truncateTagKey).setVStr(truncateTagValue))
+
     spanBuilder.build()
   }
 

commons/src/main/scala/com/expedia/www/haystack/collector/commons/config/ConfigurationLoader.scala

@@ -42,16 +42,15 @@ object ConfigurationLoader {
     * if overrides_config_path env variable exists, then we load that config file and use base conf as fallback,
     * else we load the config from env variables(prefixed with haystack) and use base conf as fallback
     *
-    * @param resourceName name of the resource file to be loaded. Default value is `config/base.conf`
+    * @param resourceName  name of the resource file to be loaded. Default value is `config/base.conf`
     * @param envNamePrefix env variable prefix to override config values. Default is `HAYSTACK_PROP_`
-    *
     * @return an instance of com.typesafe.Config
     */
-  def loadConfigFileWithEnvOverrides(resourceName : String = "config/base.conf",
-                                     envNamePrefix : String = ENV_NAME_PREFIX) : Config = {
+  def loadConfigFileWithEnvOverrides(resourceName: String = "config/base.conf",
+                                     envNamePrefix: String = ENV_NAME_PREFIX): Config = {
 
-    require(resourceName != null && resourceName.length > 0 , "resourceName is required")
-    require(envNamePrefix != null && envNamePrefix.length > 0 , "envNamePrefix is required")
+    require(resourceName != null && resourceName.length > 0, "resourceName is required")
+    require(envNamePrefix != null && envNamePrefix.length > 0, "envNamePrefix is required")
 
     val baseConfig = ConfigFactory.load(resourceName)
 
@@ -84,7 +83,7 @@ object ConfigurationLoader {
   }
 
   /**
-    *  @return new config object with haystack specific environment variables
+    * @return new config object with haystack specific environment variables
     */
   private[haystack] def parsePropertiesFromMap(envVars: Map[String, String],
                                                keysWithArrayValues: Set[String],
@@ -101,6 +100,7 @@ object ConfigurationLoader {
   /**
     * converts the env variable to HOCON format
     * for e.g. env variable HAYSTACK_KAFKA_STREAMS_NUM_STREAM_THREADS gets converted to kafka.streams.num.stream.threads
+    *
     * @param env environment variable name
     * @return variable name that complies with hocon key
     */
@@ -110,6 +110,7 @@ object ConfigurationLoader {
 
   /**
     * converts the env variable value to iterable object if it starts and ends with '[' and ']' respectively.
+    *
     * @param env environment variable value
     * @return string or iterable object
     */
@@ -153,9 +154,12 @@ object ConfigurationLoader {
       outputFormat = if (extractor.hasPath("output.format")) Format.withName(extractor.getString("output.format")) else Format.PROTO,
       spanValidation = SpanValidation(SpanMaxSize(
         maxSizeValidationConfig.getBoolean("enable"),
+        maxSizeValidationConfig.getBoolean("log.only"),
         maxSizeValidationConfig.getInt("max.size.limit"),
         maxSizeValidationConfig.getString("message.tag.key"),
-        maxSizeValidationConfig.getString("message.tag.value"))
+        maxSizeValidationConfig.getString("message.tag.value"),
+        maxSizeValidationConfig.getStringList("skip.tags").map(_.toLowerCase),
+        maxSizeValidationConfig.getStringList("skip.services").map(_.toLowerCase))
       ))
   }
 
@@ -169,7 +173,7 @@ object ConfigurationLoader {
       val props = new Properties()
       val cfg = ConfigFactory.parseMap(c._2.asInstanceOf[util.HashMap[String, Object]])
       val topic = cfg.getString("config.topic")
-      val tags =  cfg.getConfig("tags").entrySet().foldRight(Map[String, String]())((t, tMap) => {
+      val tags = cfg.getConfig("tags").entrySet().foldRight(Map[String, String]())((t, tMap) => {
         tMap + (t.getKey -> t.getValue.unwrapped().toString)
       })
       val temp = cfg.getConfig("config.props").entrySet() foreach {

commons/src/main/scala/com/expedia/www/haystack/collector/commons/config/ExtractorConfiguration.scala

@@ -28,10 +28,13 @@ object Format extends Enumeration {
 
 case class SpanValidation(spanMaxSize: SpanMaxSize)
 
-case class SpanMaxSize( enable: Boolean,
-                    maxSizeLimit: Int,
-                    infoTagKey: String,
-                    infoTagValue: String )
+case class SpanMaxSize(enable: Boolean,
+                       logOnly: Boolean,
+                       maxSizeLimit: Int,
+                       infoTagKey: String,
+                       infoTagValue: String,
+                       skipTags: Seq[String],
+                       skipServices: Seq[String])
 
 case class ExtractorConfiguration(outputFormat: Format,
                                   spanValidation: SpanValidation)

commons/src/test/scala/com/expedia/www/haystack/collector/commons/unit/KeyExtractorSpec.scala

@@ -36,7 +36,7 @@ class KeyExtractorSpec extends FunSpec with Matchers with MetricsSupport {
         "trace-id-1" -> createSpan("trace-id-1", "spanId_1", "service_1", "operation", StartTimeMicros, DurationMicros),
         "trace-id-2" -> createSpan("trace-id-2", "spanId_2", "service_2", "operation", StartTimeMicros, DurationMicros))
 
-      val spanValidationConfig = SpanValidation(SpanMaxSize(enable = false, 5000, "", ""))
+      val spanValidationConfig = SpanValidation(SpanMaxSize(enable = false, logOnly = false, 5000, "", "", Seq(), Seq()))
 
       spanMap.foreach(sp => {
         val kvPairs = new ProtoSpanExtractor(ExtractorConfiguration(Format.PROTO, spanValidationConfig), LoggerFactory.getLogger(classOf[ProtoSpanExtractor]), List()).extractKeyValuePairs(sp._2.toByteArray)
@@ -54,7 +54,7 @@ class KeyExtractorSpec extends FunSpec with Matchers with MetricsSupport {
         "trace-id-1" -> createSpan("trace-id-1", "spanId_1", "service_1", "operation", StartTimeMicros, 1),
         "trace-id-2" -> createSpan("trace-id-2", "spanId_2", "service_2", "operation", StartTimeMicros, 1))
 
-      val spanValidationConfig = SpanValidation(SpanMaxSize(enable = false, 5000, "", ""))
+      val spanValidationConfig = SpanValidation(SpanMaxSize(enable = false, logOnly = false, 5000, "", "", Seq(), Seq()))
 
       spanMap.foreach(sp => {
         val kvPairs = new ProtoSpanExtractor(ExtractorConfiguration(Format.JSON, spanValidationConfig), LoggerFactory.getLogger(classOf[ProtoSpanExtractor]), List()).extractKeyValuePairs(sp._2.toByteArray)