--- /dev/null
+#!/bin/bash
+#
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+
+IMAGE_PATH=${IMAGE_PATH:-"localhost:5000"}
+IMAGE_NAME=${IMAGE_NAME:-"clover-collector"}
+
+docker build -f docker/Dockerfile -t $IMAGE_NAME .
+docker tag $IMAGE_NAME $IMAGE_PATH/$IMAGE_NAME
+docker push $IMAGE_PATH/$IMAGE_NAME
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+from cassandra.cluster import Cluster
+from cassandra.query import BatchStatement
+import logging
+
+CASSANDRA_HOSTS = ['cassandra.default']
+
+
+class CassandraOps:
+
+ def __init__(self, hosts, port=9042, keyspace='visibility'):
+ logging.basicConfig(filename='cassops.log',
+ level=logging.DEBUG)
+ cluster = Cluster(hosts, port=port)
+ self.session = cluster.connect()
+ self.keyspace = keyspace
+
+ def truncate(self, tables=['traces', 'metrics', 'spans']):
+ self.session.set_keyspace(self.keyspace)
+ try:
+ for table in tables:
+ self.session.execute("""
+ TRUNCATE %s
+ """ % table)
+ except Exception as e:
+ logging.debug(e)
+
+ def init_visibility(self):
+ try:
+ self.session.execute("""
+ CREATE KEYSPACE %s
+ WITH replication = { 'class': 'SimpleStrategy',
+ 'replication_factor': '1' }
+ """ % self.keyspace)
+ except Exception as e:
+ logging.debug(e)
+
+ self.session.set_keyspace(self.keyspace)
+
+ try:
+ self.session.execute("""
+ CREATE TABLE IF NOT EXISTS traces (
+ traceid text,
+ processes list<text>,
+ PRIMARY KEY (traceid)
+ )
+ """)
+
+ self.session.execute("""
+ CREATE TABLE IF NOT EXISTS spans (
+ spanid text,
+ traceid text,
+ duration int,
+ start_time int,
+ processid text,
+ operation_name text,
+ node_id text,
+ http_url text,
+ upstream_cluster text,
+ PRIMARY KEY (spanid, traceid)
+ )
+ """)
+
+ self.session.execute("""
+ CREATE TABLE IF NOT EXISTS metrics (
+ m_name text,
+ m_value text,
+ m_time text,
+ service text,
+ monitor_time timestamp,
+ PRIMARY KEY (m_name, monitor_time)
+ )
+ """)
+ except Exception as e:
+ logging.debug(e)
+
+ def set_prepared(self):
+ self.session.set_keyspace(self.keyspace)
+ self.insert_tracing_stmt = self.session.prepare(
+ """
+ INSERT INTO spans (spanid, traceid, duration, operation_name,
+ node_id, http_url, upstream_cluster)
+ VALUES (?, ?, ?, ?, ?, ?, ?)
+ """
+ )
+ self.insert_metric_stmt = self.session.prepare(
+ """
+ INSERT INTO metrics
+ (m_name, m_value, m_time, service, monitor_time)
+ VALUES (?, ?, ?, ?, toTimestamp(now()))
+ """
+ )
+
+ def set_batch(self):
+ self.batch = BatchStatement()
+
+ def execute_batch(self):
+ self.session.execute(self.batch)
+
+ def insert_tracing(self, table, traceid, s, tags):
+ self.session.set_keyspace(self.keyspace)
+ if 'upstream_cluster' not in tags:
+ logging.debug('NO UPSTREAM_CLUSTER KEY')
+ tags['upstream_cluster'] = 'none'
+ try:
+ self.batch.add(self.insert_tracing_stmt,
+ (s['spanID'], traceid, s['duration'],
+ s['operationName'], tags['node_id'],
+ tags['http.url'], tags['upstream_cluster']))
+ except Exception as e:
+ logging.debug('{} {} {} {} {} {} {}'.format(s['spanID'], traceid,
+ s['duration'], s['operationName'], tags['node_id'],
+ tags['http.url'], tags['upstream_cluster']))
+ logging.debug(e)
+
+ def insert_trace(self, traceid, processes):
+ self.session.set_keyspace(self.keyspace)
+ self.session.execute(
+ """
+ INSERT INTO traces (traceid, processes)
+ VALUES (%s, %s)
+ """,
+ (traceid, processes)
+ )
+
+ def insert_metric(self, m_name, m_value, m_time, service):
+ self.session.set_keyspace(self.keyspace)
+ self.batch.add(self.insert_metric_stmt,
+ (m_name, m_value, m_time, service))
+
+
+def main():
+ cass = CassandraOps(CASSANDRA_HOSTS)
+ cass.init_visibility()
+
+
+if __name__ == '__main__':
+ main()
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+import redis
+import logging
+
+REDIS_HOST = 'redis'
+# REDIS_HOST = '10.244.0.85'
+
+
+class RedisOps:
+
+ def __init__(self, host=REDIS_HOST):
+ logging.basicConfig(filename='redisops.log',
+ level=logging.DEBUG)
+ try:
+ self.r = redis.StrictRedis(host=host, port=6379, db=0)
+ except Exception as e:
+ logging.debug(e)
+
+ def init_services(self, skey='visibility_services'):
+ service_names = ['http_lb', 'proxy_access_control']
+ for s in service_names:
+ self.r.sadd(skey, s)
+
+ def init_metrics(self, pkey='metric_prefixes', skey='metric_suffixes'):
+ metric_prefixes = ['envoy_cluster_out_', 'envoy_cluster_in_']
+ metric_suffixes = [
+ '_default_svc_cluster_local_http_internal_upstream_rq_2xx',
+ '_default_svc_cluster_local_http_upstream_cx_active']
+ for p in metric_prefixes:
+ self.r.sadd(pkey, p)
+ for s in metric_suffixes:
+ self.r.sadd(skey, s)
+
+ def get_services(self, skey='visibility_services'):
+ services = self.r.smembers(skey)
+ return services
+
+ def get_metrics(self, pkey='metric_prefixes', skey='metric_suffixes'):
+ prefixes = self.r.smembers(pkey)
+ suffixes = self.r.smembers(skey)
+ return prefixes, suffixes
+
+
+def main():
+ r = RedisOps()
+ r.init_services()
+ r.init_metrics()
+ r.get_services()
+ r.get_metrics()
+
+
+if __name__ == '__main__':
+ main()
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+FROM opnfv/clover
+
+ENV REPOS_DIR="/home/opnfv/repos"
+
+# Clover repo
+ENV CLOVER_REPO_DIR="${REPOS_DIR}/clover"
+
+# Install required python packages
+RUN python -m pip install cassandra-driver redis
+
+# Set work directory
+WORKDIR ${CLOVER_REPO_DIR}
+
+COPY /process clover/collector/process
+COPY /grpc clover/collector/grpc
+COPY /db clover/collector/db
+COPY __init__.py clover/collector/__init__.py
+
+RUN pip install .
+
+WORKDIR "${CLOVER_REPO_DIR}/clover/collector"
+
+CMD ./process/grpc_process.sh no_schema_init
--- /dev/null
+#!/bin/bash
+#
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+
+python -m grpc_tools.protoc -I./ --python_out=. --grpc_python_out=. collector.proto
--- /dev/null
+// Copyright (c) Authors of Clover
+//
+// All rights reserved. This program and the accompanying materials
+// are made available under the terms of the Apache License, Version 2.0
+// which accompanies this distribution, and is available at
+// http://www.apache.org/licenses/LICENSE-2.0
+
+syntax = "proto3";
+
+package collector;
+
+// The controller service definition.
+service Controller {
+
+ rpc StopCollector (ConfigCollector) returns (CollectorReply) {}
+ rpc StartCollector (ConfigCollector) returns (CollectorReply) {}
+ rpc InitVisibility (ConfigCassandra) returns (CollectorReply) {}
+ rpc TruncateVisibility (Schemas) returns (CollectorReply) {}
+}
+
+message ConfigCassandra {
+ string cassandra_hosts = 1;
+ int32 cassandra_port = 2;
+}
+
+message ConfigCollector {
+ string t_port = 1;
+ string t_host = 2;
+ string m_port = 3;
+ string m_host = 4;
+ string c_port = 5;
+ string c_hosts = 6;
+ string sinterval = 7;
+}
+
+message Schemas {
+ string schemas = 1;
+ string cassandra_hosts = 2;
+ int32 cassandra_port = 3;
+
+}
+
+message CollectorReply {
+ string message = 1;
+}
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+from __future__ import print_function
+from kubernetes import client, config
+
+import grpc
+import argparse
+import pickle
+
+import collector_pb2
+import collector_pb2_grpc
+
+# This is a basic client script to test server GRPC messaging
+# TODO improve interface overall
+
+
+def run(args, grpc_port='50054'):
+ pod_ip = get_podip('clover-collector')
+ if pod_ip == '':
+ return "Can not find service: {}".format(args['service_name'])
+ collector_grpc = pod_ip + ':' + grpc_port
+ channel = grpc.insecure_channel(collector_grpc)
+ stub = collector_pb2_grpc.ControllerStub(channel)
+
+ if args['cmd'] == 'init':
+ return init_visibility(stub)
+ elif args['cmd'] == 'start':
+ return start_collector(stub)
+ elif args['cmd'] == 'stop':
+ return stop_collector(stub)
+ elif args['cmd'] == 'clean':
+ return clean_visibility(stub)
+ else:
+ return "Invalid command: {}".format(args['cmd'])
+
+
+def get_podip(pod_name):
+ ip = ''
+ if pod_name != '':
+ config.load_kube_config()
+ v1 = client.CoreV1Api()
+ ret = v1.list_pod_for_all_namespaces(watch=False)
+ for i in ret.items:
+ if i.metadata.name.lower().find(pod_name.lower()) != -1:
+ print("Pod IP: {}".format(i.status.pod_ip))
+ ip = i.status.pod_ip
+ return str(ip)
+ return str(ip)
+
+
+def init_visibility(stub):
+ try:
+ cassandra_hosts = pickle.dumps(['cassandra.default'])
+ response = stub.InitVisibility(collector_pb2.ConfigCassandra(
+ cassandra_hosts=cassandra_hosts, cassandra_port=9042))
+ except Exception as e:
+ return e
+ return response.message
+
+
+def clean_visibility(stub):
+ try:
+ cassandra_hosts = pickle.dumps(['cassandra.default'])
+ schemas = pickle.dumps(['spans', 'traces', 'metrics'])
+ response = stub.TruncateVisibility(collector_pb2.Schemas(
+ schemas=schemas, cassandra_hosts=cassandra_hosts,
+ cassandra_port=9042))
+ except Exception as e:
+ return e
+ return response.message
+
+
+def start_collector(stub):
+ try:
+ cassandra_hosts = pickle.dumps(['cassandra.default'])
+ response = stub.StartCollector(collector_pb2.ConfigCollector(
+ t_port='16686', t_host='jaeger-deployment.istio-system',
+ m_port='9090', m_host='prometheus.istio-system',
+ c_port='9042', c_hosts=cassandra_hosts,
+ sinterval='5'))
+ except Exception as e:
+ return e
+ return response.message
+
+
+def stop_collector(stub):
+ try:
+ response = stub.StopCollector(collector_pb2.ConfigCollector())
+ except Exception as e:
+ return e
+ return response.message
+
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ '--cmd', required=True,
+ help='Command to execute in collector')
+ args = parser.parse_args()
+ print(run(vars(args)))
--- /dev/null
+# Generated by the protocol buffer compiler. DO NOT EDIT!
+# source: collector.proto
+
+import sys
+_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1'))
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
+from google.protobuf import symbol_database as _symbol_database
+from google.protobuf import descriptor_pb2
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+
+
+DESCRIPTOR = _descriptor.FileDescriptor(
+ name='collector.proto',
+ package='collector',
+ syntax='proto3',
+ serialized_pb=_b('\n\x0f\x63ollector.proto\x12\tcollector\"B\n\x0f\x43onfigCassandra\x12\x17\n\x0f\x63\x61ssandra_hosts\x18\x01 \x01(\t\x12\x16\n\x0e\x63\x61ssandra_port\x18\x02 \x01(\x05\"\x85\x01\n\x0f\x43onfigCollector\x12\x0e\n\x06t_port\x18\x01 \x01(\t\x12\x0e\n\x06t_host\x18\x02 \x01(\t\x12\x0e\n\x06m_port\x18\x03 \x01(\t\x12\x0e\n\x06m_host\x18\x04 \x01(\t\x12\x0e\n\x06\x63_port\x18\x05 \x01(\t\x12\x0f\n\x07\x63_hosts\x18\x06 \x01(\t\x12\x11\n\tsinterval\x18\x07 \x01(\t\"K\n\x07Schemas\x12\x0f\n\x07schemas\x18\x01 \x01(\t\x12\x17\n\x0f\x63\x61ssandra_hosts\x18\x02 \x01(\t\x12\x16\n\x0e\x63\x61ssandra_port\x18\x03 \x01(\x05\"!\n\x0e\x43ollectorReply\x12\x0f\n\x07message\x18\x01 \x01(\t2\xb3\x02\n\nController\x12H\n\rStopCollector\x12\x1a.collector.ConfigCollector\x1a\x19.collector.CollectorReply\"\x00\x12I\n\x0eStartCollector\x12\x1a.collector.ConfigCollector\x1a\x19.collector.CollectorReply\"\x00\x12I\n\x0eInitVisibility\x12\x1a.collector.ConfigCassandra\x1a\x19.collector.CollectorReply\"\x00\x12\x45\n\x12TruncateVisibility\x12\x12.collector.Schemas\x1a\x19.collector.CollectorReply\"\x00\x62\x06proto3')
+)
+
+
+
+
+_CONFIGCASSANDRA = _descriptor.Descriptor(
+ name='ConfigCassandra',
+ full_name='collector.ConfigCassandra',
+ filename=None,
+ file=DESCRIPTOR,
+ containing_type=None,
+ fields=[
+ _descriptor.FieldDescriptor(
+ name='cassandra_hosts', full_name='collector.ConfigCassandra.cassandra_hosts', index=0,
+ number=1, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='cassandra_port', full_name='collector.ConfigCassandra.cassandra_port', index=1,
+ number=2, type=5, cpp_type=1, label=1,
+ has_default_value=False, default_value=0,
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ ],
+ extensions=[
+ ],
+ nested_types=[],
+ enum_types=[
+ ],
+ options=None,
+ is_extendable=False,
+ syntax='proto3',
+ extension_ranges=[],
+ oneofs=[
+ ],
+ serialized_start=30,
+ serialized_end=96,
+)
+
+
+_CONFIGCOLLECTOR = _descriptor.Descriptor(
+ name='ConfigCollector',
+ full_name='collector.ConfigCollector',
+ filename=None,
+ file=DESCRIPTOR,
+ containing_type=None,
+ fields=[
+ _descriptor.FieldDescriptor(
+ name='t_port', full_name='collector.ConfigCollector.t_port', index=0,
+ number=1, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='t_host', full_name='collector.ConfigCollector.t_host', index=1,
+ number=2, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='m_port', full_name='collector.ConfigCollector.m_port', index=2,
+ number=3, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='m_host', full_name='collector.ConfigCollector.m_host', index=3,
+ number=4, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='c_port', full_name='collector.ConfigCollector.c_port', index=4,
+ number=5, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='c_hosts', full_name='collector.ConfigCollector.c_hosts', index=5,
+ number=6, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='sinterval', full_name='collector.ConfigCollector.sinterval', index=6,
+ number=7, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ ],
+ extensions=[
+ ],
+ nested_types=[],
+ enum_types=[
+ ],
+ options=None,
+ is_extendable=False,
+ syntax='proto3',
+ extension_ranges=[],
+ oneofs=[
+ ],
+ serialized_start=99,
+ serialized_end=232,
+)
+
+
+_SCHEMAS = _descriptor.Descriptor(
+ name='Schemas',
+ full_name='collector.Schemas',
+ filename=None,
+ file=DESCRIPTOR,
+ containing_type=None,
+ fields=[
+ _descriptor.FieldDescriptor(
+ name='schemas', full_name='collector.Schemas.schemas', index=0,
+ number=1, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='cassandra_hosts', full_name='collector.Schemas.cassandra_hosts', index=1,
+ number=2, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ _descriptor.FieldDescriptor(
+ name='cassandra_port', full_name='collector.Schemas.cassandra_port', index=2,
+ number=3, type=5, cpp_type=1, label=1,
+ has_default_value=False, default_value=0,
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ ],
+ extensions=[
+ ],
+ nested_types=[],
+ enum_types=[
+ ],
+ options=None,
+ is_extendable=False,
+ syntax='proto3',
+ extension_ranges=[],
+ oneofs=[
+ ],
+ serialized_start=234,
+ serialized_end=309,
+)
+
+
+_COLLECTORREPLY = _descriptor.Descriptor(
+ name='CollectorReply',
+ full_name='collector.CollectorReply',
+ filename=None,
+ file=DESCRIPTOR,
+ containing_type=None,
+ fields=[
+ _descriptor.FieldDescriptor(
+ name='message', full_name='collector.CollectorReply.message', index=0,
+ number=1, type=9, cpp_type=9, label=1,
+ has_default_value=False, default_value=_b("").decode('utf-8'),
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ ],
+ extensions=[
+ ],
+ nested_types=[],
+ enum_types=[
+ ],
+ options=None,
+ is_extendable=False,
+ syntax='proto3',
+ extension_ranges=[],
+ oneofs=[
+ ],
+ serialized_start=311,
+ serialized_end=344,
+)
+
+DESCRIPTOR.message_types_by_name['ConfigCassandra'] = _CONFIGCASSANDRA
+DESCRIPTOR.message_types_by_name['ConfigCollector'] = _CONFIGCOLLECTOR
+DESCRIPTOR.message_types_by_name['Schemas'] = _SCHEMAS
+DESCRIPTOR.message_types_by_name['CollectorReply'] = _COLLECTORREPLY
+_sym_db.RegisterFileDescriptor(DESCRIPTOR)
+
+ConfigCassandra = _reflection.GeneratedProtocolMessageType('ConfigCassandra', (_message.Message,), dict(
+ DESCRIPTOR = _CONFIGCASSANDRA,
+ __module__ = 'collector_pb2'
+ # @@protoc_insertion_point(class_scope:collector.ConfigCassandra)
+ ))
+_sym_db.RegisterMessage(ConfigCassandra)
+
+ConfigCollector = _reflection.GeneratedProtocolMessageType('ConfigCollector', (_message.Message,), dict(
+ DESCRIPTOR = _CONFIGCOLLECTOR,
+ __module__ = 'collector_pb2'
+ # @@protoc_insertion_point(class_scope:collector.ConfigCollector)
+ ))
+_sym_db.RegisterMessage(ConfigCollector)
+
+Schemas = _reflection.GeneratedProtocolMessageType('Schemas', (_message.Message,), dict(
+ DESCRIPTOR = _SCHEMAS,
+ __module__ = 'collector_pb2'
+ # @@protoc_insertion_point(class_scope:collector.Schemas)
+ ))
+_sym_db.RegisterMessage(Schemas)
+
+CollectorReply = _reflection.GeneratedProtocolMessageType('CollectorReply', (_message.Message,), dict(
+ DESCRIPTOR = _COLLECTORREPLY,
+ __module__ = 'collector_pb2'
+ # @@protoc_insertion_point(class_scope:collector.CollectorReply)
+ ))
+_sym_db.RegisterMessage(CollectorReply)
+
+
+
+_CONTROLLER = _descriptor.ServiceDescriptor(
+ name='Controller',
+ full_name='collector.Controller',
+ file=DESCRIPTOR,
+ index=0,
+ options=None,
+ serialized_start=347,
+ serialized_end=654,
+ methods=[
+ _descriptor.MethodDescriptor(
+ name='StopCollector',
+ full_name='collector.Controller.StopCollector',
+ index=0,
+ containing_service=None,
+ input_type=_CONFIGCOLLECTOR,
+ output_type=_COLLECTORREPLY,
+ options=None,
+ ),
+ _descriptor.MethodDescriptor(
+ name='StartCollector',
+ full_name='collector.Controller.StartCollector',
+ index=1,
+ containing_service=None,
+ input_type=_CONFIGCOLLECTOR,
+ output_type=_COLLECTORREPLY,
+ options=None,
+ ),
+ _descriptor.MethodDescriptor(
+ name='InitVisibility',
+ full_name='collector.Controller.InitVisibility',
+ index=2,
+ containing_service=None,
+ input_type=_CONFIGCASSANDRA,
+ output_type=_COLLECTORREPLY,
+ options=None,
+ ),
+ _descriptor.MethodDescriptor(
+ name='TruncateVisibility',
+ full_name='collector.Controller.TruncateVisibility',
+ index=3,
+ containing_service=None,
+ input_type=_SCHEMAS,
+ output_type=_COLLECTORREPLY,
+ options=None,
+ ),
+])
+_sym_db.RegisterServiceDescriptor(_CONTROLLER)
+
+DESCRIPTOR.services_by_name['Controller'] = _CONTROLLER
+
+# @@protoc_insertion_point(module_scope)
--- /dev/null
+# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
+import grpc
+
+import collector_pb2 as collector__pb2
+
+
+class ControllerStub(object):
+ """The controller service definition.
+ """
+
+ def __init__(self, channel):
+ """Constructor.
+
+ Args:
+ channel: A grpc.Channel.
+ """
+ self.StopCollector = channel.unary_unary(
+ '/collector.Controller/StopCollector',
+ request_serializer=collector__pb2.ConfigCollector.SerializeToString,
+ response_deserializer=collector__pb2.CollectorReply.FromString,
+ )
+ self.StartCollector = channel.unary_unary(
+ '/collector.Controller/StartCollector',
+ request_serializer=collector__pb2.ConfigCollector.SerializeToString,
+ response_deserializer=collector__pb2.CollectorReply.FromString,
+ )
+ self.InitVisibility = channel.unary_unary(
+ '/collector.Controller/InitVisibility',
+ request_serializer=collector__pb2.ConfigCassandra.SerializeToString,
+ response_deserializer=collector__pb2.CollectorReply.FromString,
+ )
+ self.TruncateVisibility = channel.unary_unary(
+ '/collector.Controller/TruncateVisibility',
+ request_serializer=collector__pb2.Schemas.SerializeToString,
+ response_deserializer=collector__pb2.CollectorReply.FromString,
+ )
+
+
+class ControllerServicer(object):
+ """The controller service definition.
+ """
+
+ def StopCollector(self, request, context):
+ # missing associated documentation comment in .proto file
+ pass
+ context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+ context.set_details('Method not implemented!')
+ raise NotImplementedError('Method not implemented!')
+
+ def StartCollector(self, request, context):
+ # missing associated documentation comment in .proto file
+ pass
+ context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+ context.set_details('Method not implemented!')
+ raise NotImplementedError('Method not implemented!')
+
+ def InitVisibility(self, request, context):
+ # missing associated documentation comment in .proto file
+ pass
+ context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+ context.set_details('Method not implemented!')
+ raise NotImplementedError('Method not implemented!')
+
+ def TruncateVisibility(self, request, context):
+ # missing associated documentation comment in .proto file
+ pass
+ context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+ context.set_details('Method not implemented!')
+ raise NotImplementedError('Method not implemented!')
+
+
+def add_ControllerServicer_to_server(servicer, server):
+ rpc_method_handlers = {
+ 'StopCollector': grpc.unary_unary_rpc_method_handler(
+ servicer.StopCollector,
+ request_deserializer=collector__pb2.ConfigCollector.FromString,
+ response_serializer=collector__pb2.CollectorReply.SerializeToString,
+ ),
+ 'StartCollector': grpc.unary_unary_rpc_method_handler(
+ servicer.StartCollector,
+ request_deserializer=collector__pb2.ConfigCollector.FromString,
+ response_serializer=collector__pb2.CollectorReply.SerializeToString,
+ ),
+ 'InitVisibility': grpc.unary_unary_rpc_method_handler(
+ servicer.InitVisibility,
+ request_deserializer=collector__pb2.ConfigCassandra.FromString,
+ response_serializer=collector__pb2.CollectorReply.SerializeToString,
+ ),
+ 'TruncateVisibility': grpc.unary_unary_rpc_method_handler(
+ servicer.TruncateVisibility,
+ request_deserializer=collector__pb2.Schemas.FromString,
+ response_serializer=collector__pb2.CollectorReply.SerializeToString,
+ ),
+ }
+ generic_handler = grpc.method_handlers_generic_handler(
+ 'collector.Controller', rpc_method_handlers)
+ server.add_generic_rpc_handlers((generic_handler,))
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+
+from concurrent import futures
+from clover.collector.db.cassops import CassandraOps
+import time
+import sys
+import grpc
+import subprocess
+import pickle
+import logging
+import collector_pb2
+import collector_pb2_grpc
+
+
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+GRPC_PORT = '[::]:50054'
+
+
+class Controller(collector_pb2_grpc.ControllerServicer):
+
+ def __init__(self, init_visibility):
+ logging.basicConfig(filename='collector_server.log',
+ level=logging.DEBUG)
+ self.collector = 0
+ if init_visibility == 'set_schemas':
+ cassandra_hosts = pickle.dumps(['cassandra.default'])
+ self.InitVisibility(collector_pb2.ConfigCassandra(
+ cassandra_port=9042, cassandra_hosts=cassandra_hosts), "")
+
+ def StopCollector(self, r, context):
+ try:
+ subprocess.Popen.kill(self.collector)
+ msg = "Stopped collector on pid: {}".format(self.collector.pid)
+ except Exception as e:
+ logging.debug(e)
+ msg = "Failed to stop collector"
+ return collector_pb2.CollectorReply(message=msg)
+
+ def StartCollector(self, r, context):
+ try:
+ self.collector = subprocess.Popen(
+ ["python", "process/collect.py",
+ "-sinterval={}".format(r.sinterval),
+ "-c_port={}".format(r.c_port),
+ "-t_port={}".format(r.t_port), "-t_host={}".format(r.t_host),
+ "-m_port={}".format(r.m_port), "-m_host={}".format(r.m_host),
+ "-c_hosts={}".format(pickle.loads(r.c_hosts)), "&"],
+ shell=False)
+ msg = "Started collector on pid: {}".format(self.collector.pid)
+ except Exception as e:
+ logging.debug(e)
+ msg = e
+ return collector_pb2.CollectorReply(message=msg)
+
+ def InitVisibility(self, r, context):
+ try:
+ cass = CassandraOps(pickle.loads(r.cassandra_hosts),
+ r.cassandra_port)
+ cass.init_visibility()
+ msg = "Added visibility schemas in cassandra"
+ except Exception as e:
+ logging.debug(e)
+ msg = "Failed to initialize cassandra"
+ return collector_pb2.CollectorReply(message=msg)
+
+ def TruncateVisibility(self, r, context):
+ try:
+ cass = CassandraOps(pickle.loads(r.cassandra_hosts),
+ r.cassandra_port)
+ cass.truncate(pickle.loads(r.schemas))
+ msg = "Truncated visibility tables"
+ except Exception as e:
+ logging.debug(e)
+ msg = "Failed to truncate visibility"
+ return collector_pb2.CollectorReply(message=msg)
+
+
+def serve(init_visibility):
+ server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+ collector_pb2_grpc.add_ControllerServicer_to_server(
+ Controller(init_visibility), server)
+ server.add_insecure_port(GRPC_PORT)
+ server.start()
+ try:
+ while True:
+ time.sleep(_ONE_DAY_IN_SECONDS)
+ except KeyboardInterrupt:
+ server.stop(0)
+
+
+if __name__ == '__main__':
+ serve(sys.argv[1])
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+from clover.tracing.tracing import Tracing
+from clover.monitoring.monitoring import Monitoring
+from clover.collector.db.cassops import CassandraOps
+from clover.collector.db.redisops import RedisOps
+
+# import pprint
+import time
+import argparse
+import logging
+import ast
+
+TRACING_SERVICES = ['istio-ingress']
+TRACING_PORT = "16686"
+MONITORING_PORT = "9090"
+CASSANDRA_PORT = 9042 # Provide as integer
+MONITORING_HOST = "prometheus.istio-system"
+TRACING_HOST = "jaeger-deployment.istio-system"
+CASSANDRA_HOSTS = ['cassandra.default']
+
+
+class Collector:
+
+ def __init__(self, t_port, t_host, m_port, m_host, c_port, c_hosts):
+ logging.basicConfig(filename='collector.log', level=logging.DEBUG)
+ try:
+ self.t = Tracing(t_host, t_port, '', False)
+ monitoring_url = "http://{}:{}".format(m_host, m_port)
+ self.m = Monitoring(monitoring_url)
+ self.c = CassandraOps(c_hosts, int(c_port))
+ self.c.set_prepared()
+ self.r = RedisOps()
+ except Exception as e:
+ logging.debug(e)
+
+ # Toplevel tracing retrieval and batch insert
+ def get_tracing(self, services, time_back=20):
+ self.c.set_batch()
+ for service in services:
+ traces = self.t.getTraces(service, time_back)
+ try:
+ self.set_tracing(traces)
+ except Exception as e:
+ logging.debug(e)
+ self.c.execute_batch()
+
+ # Insert to cassandra visibility traces and spans tables
+ def set_tracing(self, trace):
+ for traces in trace['data']:
+ for spans in traces['spans']:
+ span = {}
+ span['spanID'] = spans['spanID']
+ span['duration'] = spans['duration']
+ span['startTime'] = spans['startTime']
+ span['operationName'] = spans['operationName']
+ tag = {}
+ for tags in spans['tags']:
+ tag[tags['key']] = tags['value']
+ self.c.insert_tracing('spans', traces['traceID'],
+ span, tag)
+ process_list = []
+ for p in traces['processes']:
+ process_list.append(p)
+ service_names = []
+ for pname in process_list:
+ service_names.append(traces['processes'][pname]['serviceName'])
+ self.c.insert_trace(traces['traceID'], service_names)
+
+ # Insert to cassandra visibility metrics table
+ def get_monitoring(self):
+
+ # Fetch collector service/metric lists from redis
+ service_names = self.r.get_services()
+ metric_prefixes, metric_suffixes = self.r.get_metrics()
+
+ self.c.set_batch()
+ for sname in service_names:
+ for prefix in metric_prefixes:
+ for suffix in metric_suffixes:
+ try:
+ metric_name = prefix + sname + suffix
+ query_params = {
+ "type": "instant",
+ "query": metric_name
+ }
+ data = self.m.query(query_params)
+ m_value = data['data']['result'][0]['value'][1]
+ m_time = data['data']['result'][0]['value'][0]
+ mn = data['data']['result'][0]['metric']['__name__']
+ self.c.insert_metric(mn, m_value, str(m_time), sname)
+ except Exception as e:
+ logging.debug(e)
+ self.c.execute_batch()
+
+ # TODO add batch retrieval for monitoring metrics
+ # query_range_param = {
+ # "type": "range",
+ # "query": "tbd",
+ # "start": "60m",
+ # "end": "5m",
+ # "step": "30s"
+ # }
+ # data = self.m.query(query_range_param)
+ # pp = pprint.PrettyPrinter(indent=2)
+ # pp.pprint(data)
+
+
+def main(args):
+ if isinstance(args['c_hosts'], basestring):
+ ch = ast.literal_eval(args['c_hosts'])
+ else:
+ ch = args['c_hosts']
+
+ c = Collector(args['t_port'], args['t_host'], args['m_port'],
+ args['m_host'], args['c_port'], ch)
+
+ # Collector loop
+ loop = True
+ while loop:
+ try:
+ c.get_tracing(args['t_services'])
+ c.get_monitoring()
+ time.sleep(int(args['sinterval']))
+ except KeyboardInterrupt:
+ loop = False
+
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ '-sinterval', default=5,
+ help='Sample interval for collector loop')
+ parser.add_argument(
+ '-t_port', default=TRACING_PORT,
+ help='Port to access Jaeger tracing')
+ parser.add_argument(
+ '-m_port', default=MONITORING_PORT,
+ help='Port to access Prometheus monitoring')
+ parser.add_argument(
+ '-t_host', default=TRACING_HOST,
+ help='Host to access Jaeger tracing')
+ parser.add_argument(
+ '-m_host', default=MONITORING_HOST,
+ help='Host to access Prometheus monitoring')
+ parser.add_argument(
+ '-c_hosts', default=CASSANDRA_HOSTS,
+ help='Host(s) to access Cassandra cluster')
+ parser.add_argument(
+ '-c_port', default=CASSANDRA_PORT,
+ help='Port to access Cassandra cluster')
+ parser.add_argument(
+ '-t_services', default=TRACING_SERVICES,
+ help='Collect services on this list of services')
+
+ args, unknown = parser.parse_known_args()
+ print(main(vars(args)))
--- /dev/null
+#!/bin/bash
+#
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+
+python grpc/collector_server.py test1
--- /dev/null
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+ name: {{ deploy_name }}
+ labels:
+ app: {{ deploy_name }}
+spec:
+ template:
+ metadata:
+ labels:
+ app: {{ deploy_name }}
+ spec:
+ containers:
+ - name: {{ deploy_name }}
+ image: {{ image_path }}/{{ image_name }}:{{ image_tag }}
+ ports:
+ - containerPort: {{ grpc_port }}
+ - containerPort: {{ redis_port }}
+ - containerPort: {{ monitor_port }}
+ - containerPort: {{ trace_port }}
+ - containerPort: {{ cass_port }}
+---
+apiVersion: v1
+kind: Service
+metadata:
+ name: {{ deploy_name }}
+ labels:
+ app: {{ deploy_name }}
+spec:
+ ports:
+ - port: {{ grpc_port }}
+ name: grpc
+ - port: {{ redis_port }}
+ name: redis
+ - port: {{ trace_port }}
+ name: jaeger-deployment
+ - port: {{ monitor_port }}
+ name: prometheus
+ - port: {{ cass_port }}
+ name: cassandra
+ selector:
+ app: {{ deploy_name }}
--- /dev/null
+# Copyright (c) Authors of Clover
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+
+import argparse
+
+from jinja2 import Template
+
+
+def render_yaml(args):
+ template_file = 'manifest.template'
+ out_file = args['deploy_name'] + '.yaml'
+
+ try:
+ with open(template_file) as f:
+ tmpl = Template(f.read())
+ output = tmpl.render(
+ image_path=args['image_path'],
+ image_name=args['image_name'],
+ image_tag=args['image_tag'],
+ deploy_name=args['deploy_name'],
+ grpc_port=args['grpc_port'],
+ monitor_port=args['monitor_port'],
+ redis_port=args['redis_port'],
+ cass_port=args['cass_port'],
+ trace_port=args['trace_port']
+ )
+ with open(out_file, "wb") as fh:
+ fh.write(output)
+ return "Generated manifest for {}".format(args['deploy_name'])
+ except Exception as e:
+ print(e)
+ return "Unable to generate manifest for {}".format(
+ args['deploy_name'])
+
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ '--image_name', default='clover-collector',
+ help='The image name to use')
+ parser.add_argument(
+ # '--image_path', default='opnfv',
+ '--image_path', default='localhost:5000',
+ help='The path to the image to use')
+ parser.add_argument(
+ # '--image_tag', default='opnfv-6.0.0',
+ '--image_tag', default='latest',
+ help='The image tag to use')
+ parser.add_argument(
+ '--deploy_name', default='clover-collector',
+ help='The k8s deploy name to use')
+ parser.add_argument(
+ '--redis_port', default='6379',
+ help='The redis port to connect for management')
+ parser.add_argument(
+ '--monitor_port', default='9090',
+ help='The Prometheus monitoring port')
+ parser.add_argument(
+ '--grpc_port', default='50054',
+ help='The GRPC server port for collector management')
+ parser.add_argument(
+ '--trace_port', default='16686',
+ help='The Jaeger tracing port')
+ parser.add_argument(
+ '--cass_port', default='9042',
+ help='The Cassandra port')
+
+ args = parser.parse_args()
+ print(render_yaml(vars(args)))
print("query %s %s, status=%s, size=%d, dur=%.3f" % \
(self.host, query_params["query"], resp.status_code, len(resp.text), dur))
- pp = pprint.PrettyPrinter(indent=2)
- pp.pprint(resp.json())
+ #pp = pprint.PrettyPrinter(indent=2)
+ ##pp.pprint(resp.json())
+ return resp.json()
except Exception as e:
print("ERROR: Could not query prometheus instance %s. \n %s" % (url, e))
Code Review / clover.git / commitdiff