Create tables for oragent

create_tables.sql

-- Copyright 2020 Orijtech, Inc. All Rights Reserved.
-- Uncomment these if you have to start a fresh build.
-- DROP TABLE TracesSummary;
-- DROP TABLE MetricsSummary;
-- DROP TABLE Trace;
-- DROP TABLE Metric;
-- DROP TABLE Node;

CREATE TABLE Node (
    hash VARCHAR(32) NOT NULL PRIMARY KEY,
    pid INTEGER NOT NULL,
    start_time TIMESTAMP,
    service_name VARCHAR(1000),
    os VARCHAR(50),
    host VARCHAR(128),
    last_active_traces_time TIMESTAMP,
    last_active_metrics_time TIMESTAMP,
    language SMALLINT,
    lib_version VARCHAR(20),
    exporter_version VARCHAR(20)
);

CREATE TABLE Metric (
    id VARCHAR(32) NOT NULL PRIMARY KEY,
    name VARCHAR(1000) NOT NULL,
    descr VARCHAR(10000) NOT NULL,
    unit VARCHAR(20),
    descr_type SMALLINT NOT NULL,
    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP
);

CREATE TABLE MetricsSummary (
    metric_id VARCHAR(32) NOT NULL REFERENCES Metric(id),
    node_hash VARCHAR(32) NOT NULL REFERENCES Node(hash),
    n INTEGER NOT NULL,
    -- mag_sum represents the magnitude's sum of points 
    -- occuring at a specific point in time.
    -- mag_sum DOUBLE PRECISION NOT NULL,
    at TIMESTAMP,
    kind SMALLINT,
    PRIMARY KEY(metric_id, node_hash, at)
);

CREATE TABLE TracesSummary (
    node_hash VARCHAR(32) NOT NULL REFERENCES Node(hash),
    n INTEGER NOT NULL,
    n_with_errors INTEGER,
    at TIMESTAMP NOT NULL,
    kind SMALLINT,
    PRIMARY KEY(node_hash, at)
);

CREATE TABLE Trace (
    id VARCHAR(16) NOT NULL PRIMARY KEY,
    parent_id VARCHAR(16),
    tid VARCHAR(32) NOT NULL,
    name VARCHAR(1000) NOT NULL,
    source VARCHAR(32) NOT NULL REFERENCES Node(hash),
    start_time TIMESTAMP NOT NULL,
    end_time TIMESTAMP NOT NULL,
    saved TIMESTAMP NOT NULL
);

list_them.sql

-- SELECT * from MetricsSummary;
-- SELECT * from TracesSummary;

/* List distinct spans ordered by count */
SELECT
    name, to_char(AVG(end_time-start_time), 'HH24hMImSSsMSmxUSµs') as avg_latency, COUNT(name) as ncount
FROM
    TraceListing AS tl
WHERE
    name != ''
GROUP BY
    name
ORDER BY
    ncount DESC;

load_generator.go

package main

import (
	"bytes"
	"context"
	"fmt"
	"io/ioutil"
	"log"
	"math/rand"
	"net/http"
	"net/http/httptest"
	"time"

	"contrib.go.opencensus.io/exporter/ocagent"
	"go.opencensus.io/plugin/ochttp"
	"go.opencensus.io/stats/view"
	"go.opencensus.io/trace"
)

func main() {
	cst := httptest.NewServer(&ochttp.Handler{
		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			time.Sleep(time.Duration(rand.Intn(2500)) * time.Millisecond)
			if time.Now().UnixNano()%2 == 0 {
				http.Error(w, "Failing on purpose", 400)
			} else {
				w.Write([]byte("Bonjour"))
			}
		}),
	})
	defer cst.Close()

	allViews := append(ochttp.DefaultClientViews, ochttp.DefaultServerViews...)
	if err := view.Register(allViews...); err != nil {
		log.Fatalf("Failed to register client views for HTTP metrics: %v", err)
	}

	oce, err := ocagent.NewExporter(
		ocagent.WithInsecure(),
		ocagent.WithReconnectionPeriod(5*time.Second),
		ocagent.WithAddress("localhost:55678"), // Only included here for demo purposes.
		ocagent.WithServiceName("self_contained-example"))
	if err != nil {
		log.Fatalf("Failed to create ocagent-exporter: %v", err)
	}
	trace.RegisterExporter(oce)
	view.RegisterExporter(oce)

	// For tracing, let's always sample for the purposes of this demo
	trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})

	client := cst.Client()
	tr := client.Transport.(*http.Transport)
	client.Transport = &ochttp.Transport{Base: tr}

	for i := uint64(1); ; i++ {
		time.Sleep(time.Duration(rand.Intn(4500)) * time.Millisecond)

		ctx := context.Background()

		var req *http.Request
		if time.Now().UnixNano()%5 == 0 {
			req, _ = http.NewRequest("POST", cst.URL, bytes.NewReader([]byte("payload")))
		} else {
			req, _ = http.NewRequest("GET", cst.URL, nil)
		}

		func() {
			var err error
			spanName := spanSamples[int(time.Now().Unix())%len(spanSamples)]
			ctx, span := trace.StartSpan(ctx, spanName)
			defer func() {
				if err != nil {
					span.SetStatus(trace.Status{
						Code:    trace.StatusCodeInternal,
						Message: err.Error(),
					})
				}
				span.End()
			}()
			req = req.WithContext(ctx)
			var res *http.Response
			res, err = client.Do(req)
			if err != nil {
				fmt.Printf("Error encountered: %v", err)
				return
			}
			_, _ = ioutil.ReadAll(res.Body)
			res.Body.Close()
		}()

		fmt.Printf("Done with request: %d\r", i)
	}
}

var spanSamples = []string{
	"redis.(*Conn).readReplies",
	"(*mapbox.Client).LookupPlace",
	"golang.org/groupcache.(*Group).Get",
	"Recv.rpc.Fetch.Crawl",
	"memcache.Client.getConn",
	"memcache.Client.readReplies",
	"java.sql.Connection.executeQuery",
	"db.AddToLeaderBoard",
	"google.spanner.v1.Spanner.CreateSession",
	"cloud.google.com/go/spanner.Query",
	"Sent.google.spanner.v1.Spanner.ExecuteStreamingSql",
	"dgraph.cmd.zero.oracle.Zero.CommitOrAbort",
	"github.com/mongodb/mongo-go-driver.Database.Drop",
	"istio-ingressgateway.productpage.default.svc.cluster.localhost:9090/product",
}

load_generator.py

#!/bin/env python3

import random
import requests
import time

from opencensus.ext.ocagent import (
    stats_exporter,
    trace_exporter,
)
from opencensus.stats import aggregation as aggregation_module
from opencensus.stats import measure as measure_module
from opencensus.stats import stats as stats_module
from opencensus.stats import view as view_module
from opencensus.tags import tag_key as tag_key_module
from opencensus.tags import tag_map as tag_map_module
from opencensus.tags import tag_value as tag_value_module
from opencensus.trace import config_integration
from opencensus.trace.samplers import AlwaysOnSampler
from opencensus.trace.status import Status
from opencensus.trace.tracer import Tracer

# Create the measures
# The latency in milliseconds
m_latency_ms = measure_module.MeasureFloat("spanner/latency", "The latency in milliseconds per method", "ms")
# The stats recorder
stats_recorder = stats_module.stats.stats_recorder

key_method = tag_key_module.TagKey("method")
key_status = tag_key_module.TagKey("status")
key_error  = tag_key_module.TagKey("error")
status_OK = tag_value_module.TagValue("OK")
status_ERROR = tag_value_module.TagValue("ERROR")

tag_value_DDL = tag_value_module.TagValue("DDL")

def registerAllViews(vmgr):
    latency_view = view_module.View("method_latency", "The distribution of the latencies",
        [key_method, key_status, key_error],
        m_latency_ms,
        # Latency in buckets:
        # [>=0ms, >=25ms, >=50ms, >=75ms, >=100ms, >=200ms, >=400ms, >=600ms, >=800ms, >=1s, >=2s, >=4s, >=6s]
        aggregation_module.DistributionAggregation([1, 25, 50, 75, 100, 200, 400, 600, 800, 1000, 2000, 4000, 6000]))
    vmgr.register_view(latency_view)

def do(activity_name, tracer, fn, *args, **kwargs):
    with tracer.span(name=activity_name) as span:
        start = time.time()
        mm = stats_recorder.new_measurement_map()

        err = ''

        try:
            return fn(*args, **kwargs)
        except Exception as e:
            status_value = status_ERROR
            err = '%s' % e
            span.status = Status(5, err)
        finally:
            tm = tag_map_module.TagMap()
            tm.insert(key_method, activity_name)

            if not err:
                tm.insert(key_status, status_OK)
            else:
                tm.insert(key_error, err)
                tm.insert(key_status, status_ERROR)

            mm.measure_float_put(m_latency_ms, (time.time() - start) * 1000.0)
            mm.record(tm)

def register_all(service_name):
    view_manager = stats_module.stats.view_manager
    view_manager.register_exporter(
        stats_exporter.new_stats_exporter(
            service_name=service_name,
            endpoint='localhost:55678',
            interval=5))

    texp = trace_exporter.TraceExporter(
            service_name=service_name,
            endpoint='localhost:55678')

    view_manager.register_exporter(texp)
    registerAllViews(view_manager)
    return Tracer(exporter=texp, sampler=AlwaysOnSampler())

def sleepThenRaise():
    for i in range(1+int(random.random()*6)):
        ts = random.random() * 10
        time.sleep(ts)
        print('sleepThenRaise: %d\r' % i, end='')
    raise Exception("Done right here and there")

routes = ['/io/#Reader', '/net/http/httputil/#DumpRequest', '/non-existent', '/', '/?foo-bar']
def fetch_go_website():
    res = requests.get('https://golang.org'+random.choice(routes))
    time.sleep(random.random() * 4)
    _ = res.text

def main():
    config_integration.trace_integrations(['requests'])
    tracer = register_all('self_contained_py')
    fns = dict(sleep_then_raise=sleepThenRaise, fetch_go_website=fetch_go_website)
    while 1:
        for fnName, fn in fns.items():
            do(fnName, tracer, fn)
            time.sleep(random.random() * 7)

if __name__ == '__main__':
    main()