diff options
Diffstat (limited to 'src/main/scala/xyz/driver/core/trace')
10 files changed, 0 insertions, 287 deletions
diff --git a/src/main/scala/xyz/driver/core/trace/ExceptionLoggingSizedBufferingTraceConsumer.scala b/src/main/scala/xyz/driver/core/trace/ExceptionLoggingSizedBufferingTraceConsumer.scala deleted file mode 100644 index 399e2c6..0000000 --- a/src/main/scala/xyz/driver/core/trace/ExceptionLoggingSizedBufferingTraceConsumer.scala +++ /dev/null @@ -1,35 +0,0 @@ -package xyz.driver.core.trace - -import com.google.cloud.trace.v1.consumer.FlushableTraceConsumer -import com.google.devtools.cloudtrace.v1.Traces -import com.typesafe.scalalogging.Logger - -import scala.util.control.NonFatal - -/** - * ExceptionLoggingFlushableTraceConsumer simply wraps a flushable trace consumer and catches/logs any exceptions - * @param traceConsumer the flusable trace consumer to wrap - * @param log where to log any exceptions - */ -class ExceptionLoggingFlushableTraceConsumer(traceConsumer: FlushableTraceConsumer, log: Logger) - extends FlushableTraceConsumer { - - private val flushableTraceConsumer = traceConsumer - - private def exceptionLogger(exception: Throwable): Unit = - log.error(s"Encountered exception logging to google $exception") - - override def receive(trace: Traces): Unit = - try { - flushableTraceConsumer.receive(trace) - } catch { - case NonFatal(e) => exceptionLogger(e) - } - - override def flush(): Unit = - try { - flushableTraceConsumer.flush() - } catch { - case NonFatal(e) => exceptionLogger(e) - } -} diff --git a/src/main/scala/xyz/driver/core/trace/GoogleServiceTracer.scala b/src/main/scala/xyz/driver/core/trace/GoogleServiceTracer.scala deleted file mode 100644 index ead4c6f..0000000 --- a/src/main/scala/xyz/driver/core/trace/GoogleServiceTracer.scala +++ /dev/null @@ -1,14 +0,0 @@ -package xyz.driver.core.trace - -import akka.http.scaladsl.model.headers.RawHeader -import com.google.cloud.trace.Tracer -import com.google.cloud.trace.core.{SpanContextFactory, TraceContext} - -final case class GoogleStackdriverTraceSpan(tracer: Tracer, context: TraceContext) extends CanMakeHeader { - def header: RawHeader = - RawHeader(TracingHeaderKey, SpanContextFactory.toHeader(context.getHandle.getCurrentSpanContext)) -} - -trait GoogleServiceTracer extends ServiceTracer { - type TracerSpanPayload = GoogleStackdriverTraceSpan -} diff --git a/src/main/scala/xyz/driver/core/trace/GoogleStackdriverTrace.scala b/src/main/scala/xyz/driver/core/trace/GoogleStackdriverTrace.scala deleted file mode 100644 index fe4bb5c..0000000 --- a/src/main/scala/xyz/driver/core/trace/GoogleStackdriverTrace.scala +++ /dev/null @@ -1,56 +0,0 @@ -package xyz.driver.core.trace - -import java.io.FileInputStream -import java.nio.file.{Files, Paths} -import java.util - -import akka.http.scaladsl.model.HttpRequest -import com.google.auth.oauth2.GoogleCredentials -import com.google.cloud.trace.grpc.v1.GrpcTraceConsumer -import com.google.cloud.trace.v1.consumer.TraceConsumer -import com.typesafe.scalalogging.Logger - -final class GoogleStackdriverTrace(projectId: String, - clientSecretsFile: String, - appName: String, - appEnvironment: String, - log: Logger, - bufferSize: Int = 1024, - scheduledDelay: Int = 15) - extends GoogleServiceTracer { - - // initialize our various tracking storage systems - private val clientSecretsInputStreamOpt: Option[FileInputStream] = if (Files.exists(Paths.get(clientSecretsFile))) { - Some(new FileInputStream(clientSecretsFile)) - } else { - None - } - // if the google credentials are invalid, just log the traces - private val traceConsumer: TraceConsumer = clientSecretsInputStreamOpt.fold[TraceConsumer] { - log.warn(s"Google credentials not found in path: $clientSecretsFile") - new LoggingTraceConsumer(log) - } { clientSecretsInputStream => - GrpcTraceConsumer - .createWithCredentials( - GoogleCredentials - .fromStream(clientSecretsInputStream) - .createScoped(util.Arrays.asList("https://www.googleapis.com/auth/trace.append")) - ) - } - - private val googleServiceTracer = - new GoogleStackdriverTraceWithConsumer(projectId, - appName, - appEnvironment, - traceConsumer, - log, - bufferSize, - scheduledDelay) - - override def startSpan(httpRequest: HttpRequest): GoogleStackdriverTraceSpan = - googleServiceTracer.startSpan(httpRequest) - - override def endSpan(span: GoogleStackdriverTraceSpan): Unit = googleServiceTracer.endSpan(span) - - override def flush(): Unit = googleServiceTracer.flush() -} diff --git a/src/main/scala/xyz/driver/core/trace/GoogleStackdriverTraceWithConsumer.scala b/src/main/scala/xyz/driver/core/trace/GoogleStackdriverTraceWithConsumer.scala deleted file mode 100644 index ca1eab2..0000000 --- a/src/main/scala/xyz/driver/core/trace/GoogleStackdriverTraceWithConsumer.scala +++ /dev/null @@ -1,95 +0,0 @@ -package xyz.driver.core.trace - -import akka.http.scaladsl.model.HttpRequest -import com.google.cloud.trace.core._ -import com.google.cloud.trace.sink.TraceSink -import com.google.cloud.trace.v1.TraceSinkV1 -import com.google.cloud.trace.v1.consumer.{ScheduledBufferingTraceConsumer, TraceConsumer} -import com.google.cloud.trace.v1.producer.TraceProducer -import com.google.cloud.trace.{SpanContextHandler, SpanContextHandlerTracer, Tracer} -import com.typesafe.scalalogging.Logger -import java.util.concurrent.ScheduledThreadPoolExecutor -import java.util.concurrent.TimeUnit - -import scala.compat.java8.OptionConverters._ - -final class GoogleStackdriverTraceWithConsumer(projectId: String, - appName: String, - appEnvironment: String, - traceConsumer: TraceConsumer, - log: Logger, - bufferSize: Int, - scheduledDelay: Int) - extends GoogleServiceTracer { - - private val traceProducer: TraceProducer = new TraceProducer() - // use a UnitTraceSizer so the interpretation of bufferSize is # of spans to hold in memory prior to flushing - - // now set up the scheduled executor for time-based flushing of our tracing - // see https://goo.gl/HrPLuC - private val scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1) // this is just used for it's timing execution - scheduledThreadPoolExecutor.setKeepAliveTime(scheduledDelay.toLong, TimeUnit.SECONDS) - scheduledThreadPoolExecutor.allowCoreThreadTimeOut(true) - private val executorService = scheduledThreadPoolExecutor - private val threadSafeBufferingTraceConsumer = new ExceptionLoggingFlushableTraceConsumer( - new ScheduledBufferingTraceConsumer(traceConsumer, - new UnitTraceSizer(), - bufferSize, - scheduledDelay, - executorService), - log - ) - - private val traceSink: TraceSink = new TraceSinkV1(projectId, traceProducer, threadSafeBufferingTraceConsumer) - - private val spanContextFactory: SpanContextFactory = new SpanContextFactory( - new ConstantTraceOptionsFactory(true, true)) - private val timestampFactory: TimestampFactory = new JavaTimestampFactory() - - override def startSpan(httpRequest: HttpRequest): TracerSpanPayload = { - val parentHeaderOption: Option[akka.http.javadsl.model.HttpHeader] = - httpRequest.getHeader(TracingHeaderKey).asScala - val (spanContext: SpanContext, spanKind: SpanKind) = parentHeaderOption.fold { - (spanContextFactory.initialContext(), SpanKind.RPC_CLIENT) - } { parentHeader => - (spanContextFactory.fromHeader(parentHeader.value()), SpanKind.RPC_SERVER) - } - - val contextHandler: SpanContextHandler = new SimpleSpanContextHandler(spanContext) - val httpMethod = httpRequest.method.value - val httpHost = httpRequest.uri.authority.host.address() - val httpRelative = httpRequest.uri.toRelative.toString() - val tracer: Tracer = new SpanContextHandlerTracer(traceSink, contextHandler, spanContextFactory, timestampFactory) - // Create a span using the given timestamps. - // https://cloud.google.com/trace/docs/reference/v1/rest/v1/projects.traces#TraceSpan - val spanOptions: StartSpanOptions = (new StartSpanOptions()).setSpanKind(spanKind) - - val spanLabelBuilder = Labels - .builder() - .add("/http/method", httpMethod) - .add("/http/url", httpRelative) - .add("/http/host", httpHost) - .add("/component", appName) - .add("/environment", appEnvironment) - - parentHeaderOption.foreach { parentHeader => - spanLabelBuilder.add("/span/parent", parentHeader.value()) - } - - // The cloudTrace analysis reporting UI makes it easy to query by name prefix. - // this spanName gives us the ability to grab things that are specific to a particular UDE/env, as well as all - // endpoints in that service, as well as a particular endpoint in a particular environment/service. - val spanName: String = s"($appEnvironment->$appName)$httpRelative" - - val context: TraceContext = tracer.startSpan(spanName, spanOptions) - tracer.annotateSpan(context, spanLabelBuilder.build()) - GoogleStackdriverTraceSpan(tracer, context) - } - - override def endSpan(span: TracerSpanPayload): Unit = { - span.tracer.endSpan(span.context) - } - - override def flush(): Unit = threadSafeBufferingTraceConsumer.flush() // flush out the thread safe buffer - -} diff --git a/src/main/scala/xyz/driver/core/trace/LoggingTrace.scala b/src/main/scala/xyz/driver/core/trace/LoggingTrace.scala deleted file mode 100644 index 6a13708..0000000 --- a/src/main/scala/xyz/driver/core/trace/LoggingTrace.scala +++ /dev/null @@ -1,27 +0,0 @@ -package xyz.driver.core.trace - -import akka.http.scaladsl.model.HttpRequest -import com.google.cloud.trace.v1.consumer.TraceConsumer -import com.typesafe.scalalogging.Logger - -final class LoggingTrace(appName: String, appEnvironment: String, log: Logger, bufferSize: Int, scheduledDelay: Int) - extends GoogleServiceTracer { - - private val traceConsumer: TraceConsumer = new LoggingTraceConsumer(log) - private val googleServiceTracer = new GoogleStackdriverTraceWithConsumer( - "logging-tracer", - appName, - appEnvironment, - traceConsumer, - log, - bufferSize, - scheduledDelay - ) - - override def startSpan(httpRequest: HttpRequest): GoogleStackdriverTraceSpan = - googleServiceTracer.startSpan(httpRequest) - - override def endSpan(span: GoogleStackdriverTraceSpan): Unit = googleServiceTracer.endSpan(span) - - override def flush(): Unit = googleServiceTracer.flush() -} diff --git a/src/main/scala/xyz/driver/core/trace/LoggingTraceConsumer.scala b/src/main/scala/xyz/driver/core/trace/LoggingTraceConsumer.scala deleted file mode 100644 index 24df94a..0000000 --- a/src/main/scala/xyz/driver/core/trace/LoggingTraceConsumer.scala +++ /dev/null @@ -1,11 +0,0 @@ -package xyz.driver.core.trace - -import com.google.cloud.trace.v1.consumer.TraceConsumer -import com.google.devtools.cloudtrace.v1.Traces -import com.typesafe.scalalogging.Logger - -class LoggingTraceConsumer(log: Logger) extends TraceConsumer { - def receive(traces: Traces): Unit = { - log.trace(s"Received traces: $traces") - } -} diff --git a/src/main/scala/xyz/driver/core/trace/ServiceTracer.scala b/src/main/scala/xyz/driver/core/trace/ServiceTracer.scala deleted file mode 100644 index 1413b63..0000000 --- a/src/main/scala/xyz/driver/core/trace/ServiceTracer.scala +++ /dev/null @@ -1,19 +0,0 @@ -package xyz.driver.core.trace - -import akka.http.scaladsl.model.HttpRequest -import akka.http.scaladsl.model.headers.RawHeader - -trait CanMakeHeader { - def header: RawHeader -} - -trait ServiceTracer { - - type TracerSpanPayload <: CanMakeHeader - - def startSpan(httpRequest: HttpRequest): TracerSpanPayload - - def endSpan(span: TracerSpanPayload): Unit - - def flush(): Unit -} diff --git a/src/main/scala/xyz/driver/core/trace/SimpleSpanContextHandler.scala b/src/main/scala/xyz/driver/core/trace/SimpleSpanContextHandler.scala deleted file mode 100644 index 4ea0e8c..0000000 --- a/src/main/scala/xyz/driver/core/trace/SimpleSpanContextHandler.scala +++ /dev/null @@ -1,17 +0,0 @@ -package xyz.driver.core.trace - -import com.google.cloud.trace.SpanContextHandler -import com.google.cloud.trace.DetachedSpanContextHandle -import com.google.cloud.trace.core.{SpanContext, SpanContextHandle} - -@SuppressWarnings(Array("org.wartremover.warts.Var")) -class SimpleSpanContextHandler(rootSpan: SpanContext) extends SpanContextHandler { - private var currentSpanContext = rootSpan - - override def current(): SpanContext = currentSpanContext - - override def attach(context: SpanContext): SpanContextHandle = { - currentSpanContext = context - new DetachedSpanContextHandle(context) - } -} diff --git a/src/main/scala/xyz/driver/core/trace/UnitTraceSizer.scala b/src/main/scala/xyz/driver/core/trace/UnitTraceSizer.scala deleted file mode 100644 index a7d6032..0000000 --- a/src/main/scala/xyz/driver/core/trace/UnitTraceSizer.scala +++ /dev/null @@ -1,7 +0,0 @@ -package xyz.driver.core.trace -import com.google.cloud.trace.v1.util.Sizer -import com.google.devtools.cloudtrace.v1.Trace - -class UnitTraceSizer extends Sizer[Trace] { - override def size(sizeable: Trace) = 1 -} diff --git a/src/main/scala/xyz/driver/core/trace/package.scala b/src/main/scala/xyz/driver/core/trace/package.scala deleted file mode 100644 index 0dec833..0000000 --- a/src/main/scala/xyz/driver/core/trace/package.scala +++ /dev/null @@ -1,6 +0,0 @@ -package xyz.driver.core - -package object trace { - // this happens to be the same as the name google uses, but can be anything we want - val TracingHeaderKey: String = "X-Cloud-Trace-Context" -} |