diff --git a/libs/executors/garf/executors/entrypoints/grpc_server.py b/libs/executors/garf/executors/entrypoints/grpc_server.py index 35834f6..8ef46bb 100644 --- a/libs/executors/garf/executors/entrypoints/grpc_server.py +++ b/libs/executors/garf/executors/entrypoints/grpc_server.py @@ -16,18 +16,77 @@ import argparse import logging +import os +import subprocess +import time from concurrent import futures +import garf.executors import grpc -from garf.executors import execution_context, garf_pb2, garf_pb2_grpc, setup -from garf.executors.entrypoints.tracer import initialize_tracer +from garf.executors import ( + execution_context, + fetchers, + garf_pb2, + garf_pb2_grpc, + setup, + telemetry, + version, +) +from garf.executors.entrypoints import utils +from garf.executors.entrypoints.tracer import ( + initialize_logger, + initialize_meter, + initialize_tracer, +) from garf.executors.workflows import workflow, workflow_runner from google.protobuf.json_format import MessageToDict from grpc_reflection.v1alpha import reflection +from opentelemetry import metrics +from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer +from opentelemetry.instrumentation.logging import LoggingInstrumentor + +OTEL_SERVICE_NAME = 'garf' +LoggingInstrumentor().instrument(set_logging_format=False) + +server_start_time = time.time() + + +def _get_server_info(options): + if not (commit_sha := os.getenv('GIT_COMMIT_SHA')): + try: + commit_sha = ( + subprocess.check_output(['git', 'rev-parse', '--short', 'HEAD']) + .decode('ascii') + .strip() + ) + except Exception: + commit_sha = 'Unknown' + + yield metrics.Observation( + value=1, + attributes={ + 'version_executors': garf.executors.version.__version__, + 'version_core': garf.executors.version.core_version, + 'version_io': garf.executors.version.io_version, + 'git_commit': commit_sha, + 'server_type': 'grpc', + }, + ) + + +executor_info = telemetry.meter.create_observable_gauge( + 'garf_info', + callbacks=[_get_server_info], + unit='', + description='Build info of garf executor', +) class GarfService(garf_pb2_grpc.GarfService): def Execute(self, request, context): + telemetry.executor_requested_counter.add( + 1, attributes={'executor.source': request.source} + ) query_executor = setup.setup_executor( request.source, request.context.fetcher_parameters ) @@ -41,6 +100,10 @@ def Execute(self, request, context): return garf_pb2.ExecuteResponse(results=[result]) def ExecuteBatch(self, request, context): + n_queries = len(request.batch) + telemetry.executor_requested_counter.add( + n_queries, attributes={'executor.source': request.source} + ) query_executor = setup.setup_executor( request.source, request.context.fetcher_parameters ) @@ -73,12 +136,36 @@ def ExecuteWorkflow(self, request, context): execution_workflow = workflow.Workflow( **MessageToDict(request.workflow, preserving_proto_field_name=True) ) + telemetry.workflow_requested.add( + 1, attributes=execution_workflow.attributes + ) runner = workflow_runner.WorkflowRunner( execution_workflow=execution_workflow ) results = runner.run() return garf_pb2.ExecuteWorkflowResponse(results=results) + def GetVersion(self, request, context): + return garf_pb2.GarfVersion(version=version.__version__) + + def GetInfo(self, request, context): + return garf_pb2.GarfInfo( + executors_version=version.__version__, + core_version=version.core_version, + io_version=version.io_version, + ) + + def ListFetchers(self, request, context): + return garf_pb2.ListFetchersResponse( + results=[ + garf_pb2.FetcherInfo(name=name, version=fetcher.version) + for name, fetcher in fetchers.get_all_report_fetchers().items() + ] + ) + + def ListExecutors(self, request, context): + return garf_pb2.ListExecutorsResponse(results=setup.available_executors()) + if __name__ == '__main__': parser = argparse.ArgumentParser() @@ -88,6 +175,14 @@ def ExecuteWorkflow(self, request, context): ) args, _ = parser.parse_known_args() initialize_tracer() + meter = initialize_meter() + logger = utils.init_logging( + loglevel='INFO', logger_type='local', name=OTEL_SERVICE_NAME + ) + logger.addHandler(initialize_logger()) + + grpc_server_instrumentor = GrpcInstrumentorServer() + grpc_server_instrumentor.instrument() server = grpc.server( futures.ThreadPoolExecutor(max_workers=args.parallel_threshold) ) diff --git a/libs/executors/garf/executors/entrypoints/server.py b/libs/executors/garf/executors/entrypoints/server.py index 128c4f9..a3815f8 100644 --- a/libs/executors/garf/executors/entrypoints/server.py +++ b/libs/executors/garf/executors/entrypoints/server.py @@ -73,6 +73,7 @@ def _get_server_info(options): 'version_core': garf.executors.version.core_version, 'version_io': garf.executors.version.io_version, 'git_commit': commit_sha, + 'server_type': 'http', }, ) diff --git a/libs/executors/garf/executors/garf_pb2.py b/libs/executors/garf/executors/garf_pb2.py index 2536ebb..8ff3703 100644 --- a/libs/executors/garf/executors/garf_pb2.py +++ b/libs/executors/garf/executors/garf_pb2.py @@ -23,49 +23,60 @@ from google.protobuf import struct_pb2 as google_dot_protobuf_dot_struct__pb2 +from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\ngarf.proto\x12\x04garf\x1a\x1cgoogle/protobuf/struct.proto\"a\n\x0c\x46\x65tchRequest\x12\x0e\n\x06source\x18\x01 \x01(\t\x12\r\n\x05title\x18\x02 \x01(\t\x12\r\n\x05query\x18\x03 \x01(\t\x12#\n\x07\x63ontext\x18\x04 \x01(\x0b\x32\x12.garf.FetchContext\"G\n\rFetchResponse\x12\x0f\n\x07\x63olumns\x18\x01 \x03(\t\x12%\n\x04rows\x18\x02 \x03(\x0b\x32\x17.google.protobuf.Struct\"t\n\x0c\x46\x65tchContext\x12/\n\x10query_parameters\x18\x01 \x01(\x0b\x32\x15.garf.QueryParameters\x12\x33\n\x12\x66\x65tcher_parameters\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\"g\n\x0e\x45xecuteRequest\x12\x0e\n\x06source\x18\x01 \x01(\t\x12\r\n\x05title\x18\x02 \x01(\t\x12\r\n\x05query\x18\x03 \x01(\t\x12\'\n\x07\x63ontext\x18\x04 \x01(\x0b\x32\x16.garf.ExecutionContext\"\xbc\x01\n\x10\x45xecutionContext\x12/\n\x10query_parameters\x18\x01 \x01(\x0b\x32\x15.garf.QueryParameters\x12\x33\n\x12\x66\x65tcher_parameters\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\x12\x0e\n\x06writer\x18\x03 \x01(\t\x12\x32\n\x11writer_parameters\x18\x04 \x01(\x0b\x32\x17.google.protobuf.Struct\"d\n\x0fQueryParameters\x12&\n\x05macro\x18\x01 \x01(\x0b\x32\x17.google.protobuf.Struct\x12)\n\x08template\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\"\"\n\x0f\x45xecuteResponse\x12\x0f\n\x07results\x18\x01 \x03(\t\".\n\x0fQueryDefinition\x12\r\n\x05title\x18\x01 \x01(\t\x12\x0c\n\x04text\x18\x02 \x01(\t\"t\n\x13\x45xecuteBatchRequest\x12\x0e\n\x06source\x18\x01 \x01(\t\x12$\n\x05\x62\x61tch\x18\x02 \x03(\x0b\x32\x15.garf.QueryDefinition\x12\'\n\x07\x63ontext\x18\x03 \x01(\x0b\x32\x16.garf.ExecutionContext\"\'\n\x14\x45xecuteBatchResponse\x12\x0f\n\x07results\x18\x01 \x03(\t\"W\n\x10WorkflowMetadata\x12\x0f\n\x07version\x18\x01 \x01(\t\x12\x13\n\x0b\x64\x65scription\x18\x02 \x01(\t\x12\x1d\n\x15required_garf_version\x18\x03 \x01(\t\"\x80\x02\n\x0cWorkflowStep\x12\x0f\n\x07\x66\x65tcher\x18\x01 \x01(\t\x12\r\n\x05\x61lias\x18\x02 \x01(\t\x12&\n\x07queries\x18\x03 \x03(\x0b\x32\x15.garf.QueryDefinition\x12/\n\x10query_parameters\x18\x04 \x01(\x0b\x32\x15.garf.QueryParameters\x12\x33\n\x12\x66\x65tcher_parameters\x18\x05 \x01(\x0b\x32\x17.google.protobuf.Struct\x12\x0e\n\x06writer\x18\x06 \x01(\t\x12\x32\n\x11writer_parameters\x18\x07 \x01(\x0b\x32\x17.google.protobuf.Struct\"6\n\x0e\x43onfigMetadata\x12\x0f\n\x07version\x18\x01 \x01(\t\x12\x13\n\x0b\x64\x65scription\x18\x02 \x01(\t\"\x9b\x01\n\x06\x43onfig\x12\x0c\n\x04name\x18\x01 \x01(\t\x12&\n\x08metadata\x18\x02 \x01(\x0b\x32\x14.garf.ConfigMetadata\x12\x31\n\x11global_parameters\x18\x03 \x01(\x0b\x32\x16.garf.ExecutionContext\x12(\n\x07sources\x18\x04 \x01(\x0b\x32\x17.google.protobuf.Struct\"\x83\x01\n\x08Workflow\x12\x0c\n\x04name\x18\x01 \x01(\t\x12!\n\x05steps\x18\x02 \x03(\x0b\x32\x12.garf.WorkflowStep\x12(\n\x08metadata\x18\x03 \x01(\x0b\x32\x16.garf.WorkflowMetadata\x12\x1c\n\x06\x63onfig\x18\x04 \x01(\x0b\x32\x0c.garf.Config\"m\n\x16\x45xecuteWorkflowRequest\x12 \n\x08workflow\x18\x01 \x01(\x0b\x32\x0e.garf.Workflow\x12\x18\n\x10selected_aliases\x18\x02 \x03(\t\x12\x17\n\x0fskipped_aliases\x18\x03 \x03(\t\"*\n\x17\x45xecuteWorkflowResponse\x12\x0f\n\x07results\x18\x01 \x03(\t2\x96\x02\n\x0bGarfService\x12\x38\n\x07\x45xecute\x12\x14.garf.ExecuteRequest\x1a\x15.garf.ExecuteResponse\"\x00\x12G\n\x0c\x45xecuteBatch\x12\x19.garf.ExecuteBatchRequest\x1a\x1a.garf.ExecuteBatchResponse\"\x00\x12P\n\x0f\x45xecuteWorkflow\x12\x1c.garf.ExecuteWorkflowRequest\x1a\x1d.garf.ExecuteWorkflowResponse\"\x00\x12\x32\n\x05\x46\x65tch\x12\x12.garf.FetchRequest\x1a\x13.garf.FetchResponse\"\x00\x62\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\ngarf.proto\x12\x04garf\x1a\x1cgoogle/protobuf/struct.proto\x1a\x1bgoogle/protobuf/empty.proto\"a\n\x0c\x46\x65tchRequest\x12\x0e\n\x06source\x18\x01 \x01(\t\x12\r\n\x05title\x18\x02 \x01(\t\x12\r\n\x05query\x18\x03 \x01(\t\x12#\n\x07\x63ontext\x18\x04 \x01(\x0b\x32\x12.garf.FetchContext\"G\n\rFetchResponse\x12\x0f\n\x07\x63olumns\x18\x01 \x03(\t\x12%\n\x04rows\x18\x02 \x03(\x0b\x32\x17.google.protobuf.Struct\"t\n\x0c\x46\x65tchContext\x12/\n\x10query_parameters\x18\x01 \x01(\x0b\x32\x15.garf.QueryParameters\x12\x33\n\x12\x66\x65tcher_parameters\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\"g\n\x0e\x45xecuteRequest\x12\x0e\n\x06source\x18\x01 \x01(\t\x12\r\n\x05title\x18\x02 \x01(\t\x12\r\n\x05query\x18\x03 \x01(\t\x12\'\n\x07\x63ontext\x18\x04 \x01(\x0b\x32\x16.garf.ExecutionContext\"\xbc\x01\n\x10\x45xecutionContext\x12/\n\x10query_parameters\x18\x01 \x01(\x0b\x32\x15.garf.QueryParameters\x12\x33\n\x12\x66\x65tcher_parameters\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\x12\x0e\n\x06writer\x18\x03 \x01(\t\x12\x32\n\x11writer_parameters\x18\x04 \x01(\x0b\x32\x17.google.protobuf.Struct\"d\n\x0fQueryParameters\x12&\n\x05macro\x18\x01 \x01(\x0b\x32\x17.google.protobuf.Struct\x12)\n\x08template\x18\x02 \x01(\x0b\x32\x17.google.protobuf.Struct\"\"\n\x0f\x45xecuteResponse\x12\x0f\n\x07results\x18\x01 \x03(\t\".\n\x0fQueryDefinition\x12\r\n\x05title\x18\x01 \x01(\t\x12\x0c\n\x04text\x18\x02 \x01(\t\"t\n\x13\x45xecuteBatchRequest\x12\x0e\n\x06source\x18\x01 \x01(\t\x12$\n\x05\x62\x61tch\x18\x02 \x03(\x0b\x32\x15.garf.QueryDefinition\x12\'\n\x07\x63ontext\x18\x03 \x01(\x0b\x32\x16.garf.ExecutionContext\"\'\n\x14\x45xecuteBatchResponse\x12\x0f\n\x07results\x18\x01 \x03(\t\"W\n\x10WorkflowMetadata\x12\x0f\n\x07version\x18\x01 \x01(\t\x12\x13\n\x0b\x64\x65scription\x18\x02 \x01(\t\x12\x1d\n\x15required_garf_version\x18\x03 \x01(\t\"\x80\x02\n\x0cWorkflowStep\x12\x0f\n\x07\x66\x65tcher\x18\x01 \x01(\t\x12\r\n\x05\x61lias\x18\x02 \x01(\t\x12&\n\x07queries\x18\x03 \x03(\x0b\x32\x15.garf.QueryDefinition\x12/\n\x10query_parameters\x18\x04 \x01(\x0b\x32\x15.garf.QueryParameters\x12\x33\n\x12\x66\x65tcher_parameters\x18\x05 \x01(\x0b\x32\x17.google.protobuf.Struct\x12\x0e\n\x06writer\x18\x06 \x01(\t\x12\x32\n\x11writer_parameters\x18\x07 \x01(\x0b\x32\x17.google.protobuf.Struct\"6\n\x0e\x43onfigMetadata\x12\x0f\n\x07version\x18\x01 \x01(\t\x12\x13\n\x0b\x64\x65scription\x18\x02 \x01(\t\"\x9b\x01\n\x06\x43onfig\x12\x0c\n\x04name\x18\x01 \x01(\t\x12&\n\x08metadata\x18\x02 \x01(\x0b\x32\x14.garf.ConfigMetadata\x12\x31\n\x11global_parameters\x18\x03 \x01(\x0b\x32\x16.garf.ExecutionContext\x12(\n\x07sources\x18\x04 \x01(\x0b\x32\x17.google.protobuf.Struct\"\x83\x01\n\x08Workflow\x12\x0c\n\x04name\x18\x01 \x01(\t\x12!\n\x05steps\x18\x02 \x03(\x0b\x32\x12.garf.WorkflowStep\x12(\n\x08metadata\x18\x03 \x01(\x0b\x32\x16.garf.WorkflowMetadata\x12\x1c\n\x06\x63onfig\x18\x04 \x01(\x0b\x32\x0c.garf.Config\"m\n\x16\x45xecuteWorkflowRequest\x12 \n\x08workflow\x18\x01 \x01(\x0b\x32\x0e.garf.Workflow\x12\x18\n\x10selected_aliases\x18\x02 \x03(\t\x12\x17\n\x0fskipped_aliases\x18\x03 \x03(\t\"*\n\x17\x45xecuteWorkflowResponse\x12\x0f\n\x07results\x18\x01 \x03(\t\",\n\x0b\x46\x65tcherInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0f\n\x07version\x18\x02 \x01(\t\":\n\x14ListFetchersResponse\x12\"\n\x07results\x18\x01 \x03(\x0b\x32\x11.garf.FetcherInfo\"(\n\x15ListExecutorsResponse\x12\x0f\n\x07results\x18\x01 \x03(\t\"\x1e\n\x0bGarfVersion\x12\x0f\n\x07version\x18\x01 \x01(\t\"O\n\x08GarfInfo\x12\x19\n\x11\x65xecutors_version\x18\x01 \x01(\t\x12\x14\n\x0c\x63ore_version\x18\x02 \x01(\t\x12\x12\n\nio_version\x18\x03 \x01(\t2\x94\x04\n\x0bGarfService\x12\x38\n\x07\x45xecute\x12\x14.garf.ExecuteRequest\x1a\x15.garf.ExecuteResponse\"\x00\x12G\n\x0c\x45xecuteBatch\x12\x19.garf.ExecuteBatchRequest\x1a\x1a.garf.ExecuteBatchResponse\"\x00\x12P\n\x0f\x45xecuteWorkflow\x12\x1c.garf.ExecuteWorkflowRequest\x1a\x1d.garf.ExecuteWorkflowResponse\"\x00\x12\x32\n\x05\x46\x65tch\x12\x12.garf.FetchRequest\x1a\x13.garf.FetchResponse\"\x00\x12\x39\n\nGetVersion\x12\x16.google.protobuf.Empty\x1a\x11.garf.GarfVersion\"\x00\x12\x33\n\x07GetInfo\x12\x16.google.protobuf.Empty\x1a\x0e.garf.GarfInfo\"\x00\x12\x44\n\x0cListFetchers\x12\x16.google.protobuf.Empty\x1a\x1a.garf.ListFetchersResponse\"\x00\x12\x46\n\rListExecutors\x12\x16.google.protobuf.Empty\x1a\x1b.garf.ListExecutorsResponse\"\x00\x62\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'garf_pb2', _globals) if not _descriptor._USE_C_DESCRIPTORS: DESCRIPTOR._loaded_options = None - _globals['_FETCHREQUEST']._serialized_start=50 - _globals['_FETCHREQUEST']._serialized_end=147 - _globals['_FETCHRESPONSE']._serialized_start=149 - _globals['_FETCHRESPONSE']._serialized_end=220 - _globals['_FETCHCONTEXT']._serialized_start=222 - _globals['_FETCHCONTEXT']._serialized_end=338 - _globals['_EXECUTEREQUEST']._serialized_start=340 - _globals['_EXECUTEREQUEST']._serialized_end=443 - _globals['_EXECUTIONCONTEXT']._serialized_start=446 - _globals['_EXECUTIONCONTEXT']._serialized_end=634 - _globals['_QUERYPARAMETERS']._serialized_start=636 - _globals['_QUERYPARAMETERS']._serialized_end=736 - _globals['_EXECUTERESPONSE']._serialized_start=738 - _globals['_EXECUTERESPONSE']._serialized_end=772 - _globals['_QUERYDEFINITION']._serialized_start=774 - _globals['_QUERYDEFINITION']._serialized_end=820 - _globals['_EXECUTEBATCHREQUEST']._serialized_start=822 - _globals['_EXECUTEBATCHREQUEST']._serialized_end=938 - _globals['_EXECUTEBATCHRESPONSE']._serialized_start=940 - _globals['_EXECUTEBATCHRESPONSE']._serialized_end=979 - _globals['_WORKFLOWMETADATA']._serialized_start=981 - _globals['_WORKFLOWMETADATA']._serialized_end=1068 - _globals['_WORKFLOWSTEP']._serialized_start=1071 - _globals['_WORKFLOWSTEP']._serialized_end=1327 - _globals['_CONFIGMETADATA']._serialized_start=1329 - _globals['_CONFIGMETADATA']._serialized_end=1383 - _globals['_CONFIG']._serialized_start=1386 - _globals['_CONFIG']._serialized_end=1541 - _globals['_WORKFLOW']._serialized_start=1544 - _globals['_WORKFLOW']._serialized_end=1675 - _globals['_EXECUTEWORKFLOWREQUEST']._serialized_start=1677 - _globals['_EXECUTEWORKFLOWREQUEST']._serialized_end=1786 - _globals['_EXECUTEWORKFLOWRESPONSE']._serialized_start=1788 - _globals['_EXECUTEWORKFLOWRESPONSE']._serialized_end=1830 - _globals['_GARFSERVICE']._serialized_start=1833 - _globals['_GARFSERVICE']._serialized_end=2111 + _globals['_FETCHREQUEST']._serialized_start=79 + _globals['_FETCHREQUEST']._serialized_end=176 + _globals['_FETCHRESPONSE']._serialized_start=178 + _globals['_FETCHRESPONSE']._serialized_end=249 + _globals['_FETCHCONTEXT']._serialized_start=251 + _globals['_FETCHCONTEXT']._serialized_end=367 + _globals['_EXECUTEREQUEST']._serialized_start=369 + _globals['_EXECUTEREQUEST']._serialized_end=472 + _globals['_EXECUTIONCONTEXT']._serialized_start=475 + _globals['_EXECUTIONCONTEXT']._serialized_end=663 + _globals['_QUERYPARAMETERS']._serialized_start=665 + _globals['_QUERYPARAMETERS']._serialized_end=765 + _globals['_EXECUTERESPONSE']._serialized_start=767 + _globals['_EXECUTERESPONSE']._serialized_end=801 + _globals['_QUERYDEFINITION']._serialized_start=803 + _globals['_QUERYDEFINITION']._serialized_end=849 + _globals['_EXECUTEBATCHREQUEST']._serialized_start=851 + _globals['_EXECUTEBATCHREQUEST']._serialized_end=967 + _globals['_EXECUTEBATCHRESPONSE']._serialized_start=969 + _globals['_EXECUTEBATCHRESPONSE']._serialized_end=1008 + _globals['_WORKFLOWMETADATA']._serialized_start=1010 + _globals['_WORKFLOWMETADATA']._serialized_end=1097 + _globals['_WORKFLOWSTEP']._serialized_start=1100 + _globals['_WORKFLOWSTEP']._serialized_end=1356 + _globals['_CONFIGMETADATA']._serialized_start=1358 + _globals['_CONFIGMETADATA']._serialized_end=1412 + _globals['_CONFIG']._serialized_start=1415 + _globals['_CONFIG']._serialized_end=1570 + _globals['_WORKFLOW']._serialized_start=1573 + _globals['_WORKFLOW']._serialized_end=1704 + _globals['_EXECUTEWORKFLOWREQUEST']._serialized_start=1706 + _globals['_EXECUTEWORKFLOWREQUEST']._serialized_end=1815 + _globals['_EXECUTEWORKFLOWRESPONSE']._serialized_start=1817 + _globals['_EXECUTEWORKFLOWRESPONSE']._serialized_end=1859 + _globals['_FETCHERINFO']._serialized_start=1861 + _globals['_FETCHERINFO']._serialized_end=1905 + _globals['_LISTFETCHERSRESPONSE']._serialized_start=1907 + _globals['_LISTFETCHERSRESPONSE']._serialized_end=1965 + _globals['_LISTEXECUTORSRESPONSE']._serialized_start=1967 + _globals['_LISTEXECUTORSRESPONSE']._serialized_end=2007 + _globals['_GARFVERSION']._serialized_start=2009 + _globals['_GARFVERSION']._serialized_end=2039 + _globals['_GARFINFO']._serialized_start=2041 + _globals['_GARFINFO']._serialized_end=2120 + _globals['_GARFSERVICE']._serialized_start=2123 + _globals['_GARFSERVICE']._serialized_end=2655 # @@protoc_insertion_point(module_scope) diff --git a/libs/executors/garf/executors/garf_pb2_grpc.py b/libs/executors/garf/executors/garf_pb2_grpc.py index 22aa5c1..de3c255 100644 --- a/libs/executors/garf/executors/garf_pb2_grpc.py +++ b/libs/executors/garf/executors/garf_pb2_grpc.py @@ -4,6 +4,7 @@ import warnings from . import garf_pb2 as garf__pb2 +from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 GRPC_GENERATED_VERSION = '1.76.0' GRPC_VERSION = grpc.__version__ @@ -54,6 +55,26 @@ def __init__(self, channel): request_serializer=garf__pb2.FetchRequest.SerializeToString, response_deserializer=garf__pb2.FetchResponse.FromString, _registered_method=True) + self.GetVersion = channel.unary_unary( + '/garf.GarfService/GetVersion', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=garf__pb2.GarfVersion.FromString, + _registered_method=True) + self.GetInfo = channel.unary_unary( + '/garf.GarfService/GetInfo', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=garf__pb2.GarfInfo.FromString, + _registered_method=True) + self.ListFetchers = channel.unary_unary( + '/garf.GarfService/ListFetchers', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=garf__pb2.ListFetchersResponse.FromString, + _registered_method=True) + self.ListExecutors = channel.unary_unary( + '/garf.GarfService/ListExecutors', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=garf__pb2.ListExecutorsResponse.FromString, + _registered_method=True) class GarfServiceServicer(object): @@ -83,6 +104,30 @@ def Fetch(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def GetVersion(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def GetInfo(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def ListFetchers(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def ListExecutors(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def add_GarfServiceServicer_to_server(servicer, server): rpc_method_handlers = { @@ -106,6 +151,26 @@ def add_GarfServiceServicer_to_server(servicer, server): request_deserializer=garf__pb2.FetchRequest.FromString, response_serializer=garf__pb2.FetchResponse.SerializeToString, ), + 'GetVersion': grpc.unary_unary_rpc_method_handler( + servicer.GetVersion, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=garf__pb2.GarfVersion.SerializeToString, + ), + 'GetInfo': grpc.unary_unary_rpc_method_handler( + servicer.GetInfo, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=garf__pb2.GarfInfo.SerializeToString, + ), + 'ListFetchers': grpc.unary_unary_rpc_method_handler( + servicer.ListFetchers, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=garf__pb2.ListFetchersResponse.SerializeToString, + ), + 'ListExecutors': grpc.unary_unary_rpc_method_handler( + servicer.ListExecutors, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=garf__pb2.ListExecutorsResponse.SerializeToString, + ), } generic_handler = grpc.method_handlers_generic_handler( 'garf.GarfService', rpc_method_handlers) @@ -224,3 +289,111 @@ def Fetch(request, timeout, metadata, _registered_method=True) + + @staticmethod + def GetVersion(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/garf.GarfService/GetVersion', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + garf__pb2.GarfVersion.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def GetInfo(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/garf.GarfService/GetInfo', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + garf__pb2.GarfInfo.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def ListFetchers(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/garf.GarfService/ListFetchers', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + garf__pb2.ListFetchersResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def ListExecutors(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary( + request, + target, + '/garf.GarfService/ListExecutors', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + garf__pb2.ListExecutorsResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) diff --git a/libs/executors/pyproject.toml b/libs/executors/pyproject.toml index 11d9699..b93bd34 100644 --- a/libs/executors/pyproject.toml +++ b/libs/executors/pyproject.toml @@ -60,6 +60,7 @@ server=[ "fastapi[standard]", "pydantic-settings", "opentelemetry-instrumentation-fastapi", + "opentelemetry-instrumentation-grpc", "typer", "grpcio-reflection", "celery[redis]", diff --git a/libs/executors/tests/end-to-end/test_grpc_server.py b/libs/executors/tests/end-to-end/test_grpc_server.py index 5b051d6..aa28ca8 100644 --- a/libs/executors/tests/end-to-end/test_grpc_server.py +++ b/libs/executors/tests/end-to-end/test_grpc_server.py @@ -14,9 +14,10 @@ import pathlib import pytest +from garf.executors import fetchers, garf_pb2_grpc, setup, version from garf.executors import garf_pb2 as pb -from garf.executors import garf_pb2_grpc from garf.executors.entrypoints import grpc_server +from google.protobuf import empty_pb2 from google.protobuf.json_format import MessageToDict _SCRIPT_PATH = pathlib.Path(__file__).parent @@ -145,3 +146,37 @@ def test_execute_workflow(grpc_stub): ) result = grpc_stub.ExecuteWorkflow(request) assert '1-fake-test' in result.results[0] + + +def test_garf_version(grpc_stub): + result = grpc_stub.GetVersion(empty_pb2.Empty()) + assert result.version == version.__version__ + + +def test_garf_info(grpc_stub): + result = grpc_stub.GetInfo(empty_pb2.Empty()) + expected_response = pb.GarfInfo( + executors_version=version.__version__, + core_version=version.core_version, + io_version=version.io_version, + ) + assert result == expected_response + + +def test_list_fetchers(grpc_stub): + result = grpc_stub.ListFetchers(empty_pb2.Empty()) + expected_response = pb.ListFetchersResponse( + results=[ + pb.FetcherInfo(name=name, version=fetcher.version) + for name, fetcher in fetchers.get_all_report_fetchers().items() + ] + ) + assert result == expected_response + + +def test_list_executors(grpc_stub): + result = grpc_stub.ListExecutors(empty_pb2.Empty()) + expected_response = pb.ListExecutorsResponse( + results=setup.available_executors() + ) + assert result == expected_response diff --git a/protos/garf.proto b/protos/garf.proto index e16ad80..debe857 100644 --- a/protos/garf.proto +++ b/protos/garf.proto @@ -1,6 +1,7 @@ syntax = "proto3"; package garf; import "google/protobuf/struct.proto"; +import "google/protobuf/empty.proto"; service GarfService { @@ -8,7 +9,10 @@ service GarfService { rpc ExecuteBatch(ExecuteBatchRequest) returns (ExecuteBatchResponse) {} rpc ExecuteWorkflow(ExecuteWorkflowRequest) returns (ExecuteWorkflowResponse) {} rpc Fetch(FetchRequest) returns (FetchResponse) {} - + rpc GetVersion(google.protobuf.Empty) returns (GarfVersion) {} + rpc GetInfo(google.protobuf.Empty) returns (GarfInfo) {} + rpc ListFetchers(google.protobuf.Empty) returns (ListFetchersResponse) {} + rpc ListExecutors(google.protobuf.Empty) returns (ListExecutorsResponse) {} } message FetchRequest { @@ -114,3 +118,28 @@ message ExecuteWorkflowRequest { message ExecuteWorkflowResponse { repeated string results = 1; } + + +// List +message FetcherInfo { + string name = 1; + string version = 2; +} +message ListFetchersResponse { + repeated FetcherInfo results = 1; +} + +message ListExecutorsResponse { + repeated string results = 1; +} + +// Info +message GarfVersion { + string version = 1; +} + +message GarfInfo { + string executors_version = 1; + string core_version = 2; + string io_version = 3; +}