Distributed Tracing for Functions

#
# oci-invoke-function-python version 2.0.
#
# Copyright (c) 2021 Oracle, Inc.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl.
#

import io
import json
import logging
import oci
from fdk import response

def handler(ctx, data: io.BytesIO=None):
    app_name = ctx.AppName()
    func_name = ctx.FnName()
    logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler")

    try:
        body = json.loads(data.getvalue())
        function_endpoint = body.get("function_endpoint")
        function_ocid = body.get("function_ocid")
        function_body = body.get("function_body")
    except (Exception) as ex:
        print('ERROR: Missing key in payload', ex, flush=True)
        raise
    
    signer = oci.auth.signers.get_resource_principals_signer()
    client = oci.functions.FunctionsInvokeClient(config={}, signer=signer, service_endpoint=function_endpoint)
    
    #
    # Zipkin X-B3- header propagation
    #
    tracing_context = ctx.TracingContext()
    trace_id = tracing_context.trace_id()
    span_id = tracing_context.span_id()
    parent_span_id = tracing_context.parent_span_id()
    is_sampled = tracing_context.is_sampled()

    # if tracing is enabled, is_sampled will be true in the tracing context
    if is_sampled:
        # To propagate headers in the OCI SDK in the request to the next function,
        # add the X-B3- headers in the request. This header will be included in ALL
        # subsequent calls made.
        if trace_id is not None:
            client.base_client.session.headers['X-B3-TraceId'] = trace_id
            logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler | trace_id: " + trace_id)
        if span_id is not None:
            client.base_client.session.headers['X-B3-SpanId'] = span_id
            logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler | span_id: " + span_id)
        if parent_span_id is not None:
            client.base_client.session.headers['X-B3-ParentSpanId'] = parent_span_id
            logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler | parent_span_id: " + parent_span_id)
        client.base_client.session.headers['X-B3-Sampled'] = str(int(is_sampled))
        logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler | is_sampled: " + str(int(is_sampled)))
    else:
        # function.trace is DISABLED
        logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler | function tracing is DISABLED")

    resp = client.invoke_function(function_id=function_ocid, invoke_function_body=function_body)
    logging.getLogger().info("Inside app: " + app_name + " | function: " + func_name + " | method: handler | Response: " + resp.data.text)

    return response.Response(
        ctx, 
        response_data=resp.data.text,
        headers={"Content-Type": "application/json"}
    )
            

package com.example.fn;
import brave.Span;
import brave.Tracer;
import brave.Tracing;
import brave.propagation.*;
import brave.sampler.Sampler;
import com.fnproject.fn.api.tracing.TracingContext;
import com.github.kristofa.brave.IdConversion;
import zipkin2.reporter.Sender;
import zipkin2.reporter.brave.AsyncZipkinSpanHandler;
import zipkin2.reporter.urlconnection.URLConnectionSender;

public class HelloFunction {
    Sender sender;
    AsyncZipkinSpanHandler zipkinSpanHandler;
    Tracing tracing;
    Tracer tracer;
    String apmUrl;
    TraceContext traceContext;
    public void intializeZipkin(TracingContext tracingContext) throws Exception {
        System.out.println("Initializing the variables");
        apmUrl = tracingContext.getTraceCollectorURL();
        sender = URLConnectionSender.create(apmUrl);
        zipkinSpanHandler = AsyncZipkinSpanHandler.create(sender);
        tracing = Tracing.newBuilder()
                .localServiceName(tracingContext.getServiceName())
                .sampler(Sampler.NEVER_SAMPLE)
                .addSpanHandler(zipkinSpanHandler)
                .build();
        tracer = tracing.tracer();
        tracing.setNoop(!tracingContext.isTracingEnabled());
        traceContext = TraceContext.newBuilder()
                .traceId(IdConversion.convertToLong(tracingContext.getTraceId()))
                .spanId(IdConversion.convertToLong(tracingContext.getSpanId()))
                .sampled(tracingContext.isSampled()).build();
    }

    public String handleRequest(String input, TracingContext tracingContext) {
        try {
            intializeZipkin(tracingContext);
            // Start a new trace or a span within an existing trace representing an operation
            Span span = tracer.newChild(traceContext).name("MainHandle").start();
            System.out.println("Inside Java Hello World function");
            try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
                method1();
                method2();
                method3();
            } catch (RuntimeException | Error e) {
                span.error(e); // Unless you handle exceptions, you might not know the operation failed!
                throw e;
            } finally {
                span.finish(); // note the scope is independent of the span. Always finish a span.
                tracing.close();
                zipkinSpanHandler.flush();
            }
        } catch (Exception e) {
            return e.getMessage();
        }
        return "Hello, AppName " + tracingContext.getAppName() + " :: fnName " + tracingContext.getFunctionName();
    }

    public void method1() {
        System.out.println("Inside Method1 function");
        TraceContext traceContext = tracing.currentTraceContext().get();
        Span span = tracer.newChild(traceContext).name("Method1").start();
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            span.finish();
        }
    }

    public void method2() {
        System.out.println("Inside Method2 function");
        TraceContext traceContext = tracing.currentTraceContext().get();
        Span span = tracer.newChild(traceContext).name("Method2").start();
        try {
            Thread.sleep(400);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            span.finish();
        }
    }

    public void method3() {
        System.out.println("Inside Method3 function");
        TraceContext traceContext = tracing.currentTraceContext().get();
        Span span = tracer.newChild(traceContext).name("Method3").start();
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            span.finish();
        }
    }
}
            

import io
import json
import logging

from fdk import response

import requests
import time
from py_zipkin import Encoding
from py_zipkin.zipkin import zipkin_span
from collections import namedtuple


# transport handler, needed by py_zipkin
def transport_handler(encoded_span, tracing_context):
    return requests.post(
        tracing_context.trace_collector_url(),
        data=encoded_span,
        headers={"Content-Type": "application/json"},
    )


def handler(ctx, data: io.BytesIO = None):
    tracing_context = ctx.TracingContext()
    with zipkin_span(
        service_name=tracing_context.service_name(),
        span_name="Customer Code",
        transport_handler=(
            lambda encoded_span: transport_handler(
                encoded_span, tracing_context
            )
        ),
        zipkin_attrs=tracing_context.zipkin_attrs(),
        encoding=Encoding.V2_JSON,
        binary_annotations=tracing_context.annotations()
    ):
        name = "World"
        try:
            body = json.loads(data.getvalue())
            name = body.get("name")
        except (Exception, ValueError) as ex:
            logging.getLogger().info('error parsing json payload: ' + str(ex))

        logging.getLogger().info("Inside Python Hello World function")
        time.sleep(0.005)
        example(ctx)
        return response.Response(
            ctx, response_data=json.dumps(
                {"message": "Hello {0}".format(name)}),
            headers={"Content-Type": "application/json"}
        )


def example(ctx):
    with zipkin_span(
        service_name=ctx.TracingContext().service_name(),
        span_name="Get ADB Password from OCI Vault",
        binary_annotations=ctx.TracingContext().annotations()
    ) as example_span_context:
        try:
            logging.getLogger().debug("Get ADB Password from OCI Vault")
            time.sleep(0.005)
            # throwing an exception to show how to add error messages to spans
            raise Exception('Request failed')
        except (Exception, ValueError) as error:
            example_span_context.update_binary_annotations(
                {"Error": True, "errorMessage": str(error)}
            )
        else:
            FakeResponse = namedtuple("FakeResponse", "status, message")
            fakeResponse = FakeResponse(200, "OK")
            # how to update the span dimensions/annotations
            example_span_context.update_binary_annotations(
                {
                    "responseCode": fakeResponse.status,
                    "responseMessage": fakeResponse.message
                }
            )
            

const fdk = require('@fnproject/fdk')

// ZipkinJS core components.
const {
  ExplicitContext,
  Tracer,
  TraceId,
  BatchRecorder,
  jsonEncoder,
  sampler,
  option
} = require('zipkin')

// An HTTP transport for dispatching Zipkin traces.
const { HttpLogger } = require('zipkin-transport-http')

fdk.handle(async function (input, ctx) {
  var tracer = createOCITracer(ctx)

  var result
  // Start a new 'scoped' server handling span.
  await tracer.scoped(async function () {
    // Fetch some resource
    result = await tracer.local('fetchResource', () => {
      return fetchResource()
    })
    // Perform some processing
    result = await tracer.local('processResource', () => {
      return someComputation(result)
    })
    // Update some resource
    result = await tracer.local('updateResource', () => {
      return updateResource(result)
    })
    await flush()
  })

  return result
})

// ----------------------------------------------------------------------------
// App Simulation Functions
//

/**
 * Simulate fetching some required resource. This could be another OCI service
 * or an external call.
 *
 * @returns A Promise with the success or failure of the operation.
 */
function fetchResource () {
  return simulate(1000, { fetchResource: 'OK' })
}

/**
 * Simulate some work. This could be another OCI service.
 *
 * @returns A Promise with the success or failure of the operation.
 */
async function someComputation (toReturn) {
  var i
  for (i = 0; i < 5; i++) {
    await simulate(1000)
  }
  toReturn.processResource = 'OK'
  return toReturn
}

/**
 * Simulate updating some resource. This could be another OCI service or an
 * external call.
 *
 * @returns A Promise with the success or failure of the operation.
 */
async function updateResource (toReturn) {
  await simulate(500)
  toReturn.updateResource = 'OK'
  return toReturn
}

/**
 * A helper function to simulate an operation that takes a specified amount of time.
 *
 * @param {*} ms The simulated time for the activity in milliseconds.
 * @returns      A promise that resolves when the simulated activity finishes.
 */
function simulate (ms, result) {
  return new Promise(resolve => setTimeout(resolve, ms, result))
}

/**
 * Functions service may freeze or terminate the container on completion.
 * This function gives extra time to allow the runtime to flush any pending traces.
 * See: https://github.com/openzipkin/zipkin-js/issues/507
 *
 * @returns A Promise to await on.
 */
function flush () {
  return new Promise(resolve => setTimeout(resolve, 1000))
}

// ----------------------------------------------------------------------------
// OpenZipkin ZipkinJS Utility Functions
//

/**
 * Creates a basic Zipkin Tracer using values from context of the function
 * invocation.
 *
 * @param {*} ctx The function invocation context.
 * @returns       A configured Tracer for automatically tracing calls.
 */
function createOCITracer (ctx) {
  // An OCI APM configured Tracer
  //
  const tracingCxt = ctx.tracingContext
  const tracer = new Tracer({
    ctxImpl: new ExplicitContext(),
    recorder: new BatchRecorder({
      logger: new HttpLogger({
        // The configured OCI APM endpoint is available in the function
        // invocation context.
        endpoint: tracingCxt.traceCollectorUrl,
        jsonEncoder: jsonEncoder.JSON_V2
      })
    }),
    // APM Dimensions that should be included in all traces can be configured
    // directly on Tracer.
    defaultTags: createOCITags(ctx),
    // A custom sampling strategy can be defined.
    sampler: createOCISampler(ctx),
    localServiceName: tracingCxt.serviceName,
    supportsJoin: true,
    traceId128Bit: true
  })

  // The initial function invocation trace identifiers can be added directly.
  // If this is not defined a default TraceId is created.
  const traceId = createOCITraceId(tracer, ctx)
  tracer.setId(traceId)
  return tracer
}

/**
 * A ZipkinJS 'TraceId' can be created directly from the function invocation
 * context.
 *
 * @param {*} ctx The function invocation context.
 * @returns       A ZipkinJS 'TraceId' created from the invocation context.
 */
function createOCITraceId (tracer, ctx) {
  const tracingCxt = ctx.tracingContext
  if (tracingCxt.traceId && tracingCxt.spanId) {
    return new TraceId({
      traceId: tracingCxt.traceId,
      spanId: tracingCxt.spanId,
      sampled: new option.Some(tracingCxt.sampled),
      debug: new option.Some(tracingCxt.debug),
      shared: false
    })
  } else {
    return tracer.createRootId(
      new option.Some(tracingCxt.sampled),
      new option.Some(tracingCxt.debug)
    )
  }
}

/**
 * A ZipkinJS 'TraceId' can be crated directly from the function invocation
 * context.
 *
 * This configurations will automatically add the function meta-data as APM
 * dimensions to each trace. Function environment variable and other dimensions
 * could also be added.
 *
 * @param {*} ctx The function invocation context.
 * @returns       A map of key-value pairs, that will be added as APM
 *                dimensions to the traces.
 */
function createOCITags (ctx) {
  return {
    appID: ctx.appID,
    appName: ctx.appName,
    fnID: ctx.fnID,
    fnName: ctx.fnName
  }
}

/**
 * A ZipkinJS 'Sampler' can be created directly from the function invocation
 * context.
 *
 * This configuration will only create a trace if the function is configured
 * for tracing.
 *
 * @param {*} ctx The function invocation context.
 * @returns       A ZipkinJS 'TraceId' created from the invocation context.
 */
function createOCISampler (ctx) {
  return new sampler.Sampler((traceId) => ctx.tracingContext.isEnabled)
}