From 5b6fdf91b9a1629130fb0ddf7b2e5ea46e5f8a54 Mon Sep 17 00:00:00 2001 From: Andrei Markin Date: Mon, 20 Jul 2026 12:35:23 +0400 Subject: [PATCH] feat(executors): Add support for grpc remote execution via CLI * Ensure that all `grf` commands can be sent to remote server either via `http` or `grpc` protocol * Add new `server_type` CLI flag * Instrument grpc_server for OTEL * Add healthcheck for grpc_server --- .../garf/executors/entrypoints/grpc_server.py | 30 ++- .../garf/executors/entrypoints/typer_cli.py | 221 +++++++++++++----- libs/executors/garf/executors/garf_pb2.py | 72 +++--- .../garf/executors/workflows/workflow.py | 10 +- libs/executors/pyproject.toml | 1 + protos/garf.proto | 16 +- 6 files changed, 247 insertions(+), 103 deletions(-) diff --git a/libs/executors/garf/executors/entrypoints/grpc_server.py b/libs/executors/garf/executors/entrypoints/grpc_server.py index 8ef46bb..bd50c85 100644 --- a/libs/executors/garf/executors/entrypoints/grpc_server.py +++ b/libs/executors/garf/executors/entrypoints/grpc_server.py @@ -40,6 +40,7 @@ ) from garf.executors.workflows import workflow, workflow_runner from google.protobuf.json_format import MessageToDict +from grpc_health.v1 import health_pb2, health_pb2_grpc from grpc_reflection.v1alpha import reflection from opentelemetry import metrics from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer @@ -47,6 +48,9 @@ OTEL_SERVICE_NAME = 'garf' LoggingInstrumentor().instrument(set_logging_format=False) +grpc_server_instrumentor = GrpcInstrumentorServer() +grpc_server_instrumentor.instrument() + server_start_time = time.time() @@ -166,6 +170,16 @@ def ListFetchers(self, request, context): def ListExecutors(self, request, context): return garf_pb2.ListExecutorsResponse(results=setup.available_executors()) + def Check(self, request, context): + return health_pb2.HealthCheckResponse( + status=health_pb2.HealthCheckResponse.SERVING + ) + + def Watch(self, request, context): + return health_pb2.HealthCheckResponse( + status=health_pb2.HealthCheckResponse.UNIMPLEMENTED + ) + if __name__ == '__main__': parser = argparse.ArgumentParser() @@ -174,21 +188,23 @@ def ListExecutors(self, request, context): '--parallel-threshold', dest='parallel_threshold', default=10, type=int ) args, _ = parser.parse_known_args() - initialize_tracer() - meter = initialize_meter() + otel_service_name = os.getenv('OTEL_SERVICE_NAME', OTEL_SERVICE_NAME) + initialize_tracer(otel_service_name) + meter = initialize_meter(otel_service_name) logger = utils.init_logging( - loglevel='INFO', logger_type='local', name=OTEL_SERVICE_NAME + 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) + futures.ThreadPoolExecutor(max_workers=args.parallel_threshold), ) service = GarfService() garf_pb2_grpc.add_GarfServiceServicer_to_server(service, server) + health_pb2_grpc.add_HealthServicer_to_server(service, server) SERVICE_NAMES = ( garf_pb2.DESCRIPTOR.services_by_name['GarfService'].full_name, reflection.SERVICE_NAME, @@ -196,5 +212,5 @@ def ListExecutors(self, request, context): reflection.enable_server_reflection(SERVICE_NAMES, server) server.add_insecure_port(f'[::]:{args.port}') server.start() - logging.info('Garf service started, listening on port %d', 50051) + logging.info('Garf service started, listening on port %d', args.port) server.wait_for_termination() diff --git a/libs/executors/garf/executors/entrypoints/typer_cli.py b/libs/executors/garf/executors/entrypoints/typer_cli.py index f0fd08c..77bd688 100644 --- a/libs/executors/garf/executors/entrypoints/typer_cli.py +++ b/libs/executors/garf/executors/entrypoints/typer_cli.py @@ -21,10 +21,12 @@ from typing import Optional import garf.executors +import garf.executors.garf_pb2 as pb +import grpc import requests import typer from garf.core import cache -from garf.executors import exceptions, setup +from garf.executors import exceptions, garf_pb2_grpc, setup from garf.executors.config import Config from garf.executors.entrypoints import utils from garf.executors.entrypoints.tracer import ( @@ -34,7 +36,9 @@ from garf.executors.telemetry import tracer from garf.executors.workflows import workflow, workflow_runner from garf.io import reader, writer +from google.protobuf.json_format import ParseDict from opentelemetry import trace +from opentelemetry.instrumentation.grpc import GrpcInstrumentorClient from opentelemetry.instrumentation.logging import LoggingInstrumentor from opentelemetry.trace.propagation.tracecontext import ( TraceContextTextMapPropagator, @@ -44,6 +48,8 @@ from typing_extensions import Annotated LoggingInstrumentor().instrument(set_logging_format=False) +client_instrumentor = GrpcInstrumentorClient() +client_instrumentor.instrument() console = Console() @@ -52,6 +58,10 @@ ((f, f) for f in sorted(setup.find_executors())), ) +ServerTypeEnum = enum.Enum( + 'ServerTypeEnum', (('http', 'http'), ('grpc', 'grpc')) +) + initialize_tracer() typer_app = typer.Typer( help='Garf\n\nCall APIs with SQL in your terminal', rich_markup_mode='rich' @@ -154,16 +164,21 @@ def prune(time_days: int): ] -def _init_runner( - file, context=None, config=None -) -> workflow_runner.WorkflowRunner: - wf_parent = pathlib.Path.cwd() / pathlib.Path(file).parent +def _init_workflow(file, context=None, config=None) -> workflow.Workflow: execution_workflow = workflow.Workflow.from_file( path=file, context=context, config_file=config ) garf.executors.version.validate_version( execution_workflow.metadata.required_garf_version ) + return execution_workflow + + +def _init_runner( + file, context=None, config=None +) -> workflow_runner.WorkflowRunner: + execution_workflow = _init_workflow(file, context, config) + wf_parent = pathlib.Path.cwd() / pathlib.Path(file).parent return workflow_runner.WorkflowRunner( execution_workflow=execution_workflow, wf_parent=wf_parent ) @@ -203,6 +218,9 @@ def execute( server_url: Annotated[ str | None, typer.Option(help='Address of garf server in HOST:PORT format') ] = None, + server_type: Annotated[ + ServerTypeEnum, typer.Option(help='Type of server') + ] = 'http', additional_output: Annotated[ Optional[str], typer.Option( @@ -275,46 +293,10 @@ def execute( with tracer.start_as_current_span('read_queries'): batch = {query: reader_client.read(query) for query in found_queries} if server_url: - rest_context = context.model_dump() - if rest_context.get('writer') == ['console']: - del rest_context['writer'] - headers = {} - TraceContextTextMapPropagator().inject(headers) - if parallel_queries and len(batch) > 1: - endpoint = f'{server_url}/api/execute:batch' - request = { - 'batch': batch, - 'source': source, - 'context': rest_context, - } - try: - response = requests.post(url=endpoint, json=request, headers=headers) - response.raise_for_status() - except requests.exceptions.HTTPError as e: - raise exceptions.GarfExecutorError( - f'Server error: {e.response.status_code} - {e.response.json()}' - ) from e - - with tracer.start_as_current_span('parse_results batch'): - typer.secho(response.json().get('results')) - else: - endpoint = f'{server_url}/api/execute' - for title, text in batch.items(): - request = { - 'source': source, - 'title': title, - 'query': text, - 'context': rest_context, - } - try: - response = requests.post(url=endpoint, json=request, headers=headers) - response.raise_for_status() - except requests.exceptions.HTTPError as e: - raise exceptions.GarfExecutorError( - f'Server error: {e.response.status_code} - {e.response.json()}' - ) from e - with tracer.start_as_current_span(f'parse_results {title}'): - typer.secho(response.json().get('results')) + if server_type == ServerTypeEnum.http: + _send_http(context, parallel_queries, batch, server_url, source) + elif server_type == ServerTypeEnum.grpc: + _send_grpc(context, parallel_queries, batch, server_url, source) else: query_executor = setup.setup_executor( source=source, @@ -363,6 +345,12 @@ def run( enable_cache: EnableCache = False, cache_ttl_seconds: CacheTTL = 3600, simulate: Simulate = False, + server_url: Annotated[ + str | None, typer.Option(help='Address of garf server in HOST:PORT format') + ] = None, + server_type: Annotated[ + ServerTypeEnum, typer.Option(help='Type of server') + ] = 'http', ): """Runs workflow from a file.""" span = trace.get_current_span() @@ -373,14 +361,21 @@ def run( ) garf_logger.addHandler(initialize_logger()) context = utils.ParamsParser().parse_all(ctx.args) - runner = _init_runner(file, context, config) - runner.run( - enable_cache=enable_cache, - cache_ttl_seconds=cache_ttl_seconds, - selected_aliases=include, - skipped_aliases=exclude, - simulate=simulate, - ) + if server_url: + if server_type == ServerTypeEnum.http: + _run_worklow_http(server_url, file) + elif server_type == ServerTypeEnum.grpc: + execution_workflow = _init_workflow(file, context, config) + _run_worklow_grpc(server_url, execution_workflow) + else: + runner = _init_runner(file, context, config) + runner.run( + enable_cache=enable_cache, + cache_ttl_seconds=cache_ttl_seconds, + selected_aliases=include, + skipped_aliases=exclude, + simulate=simulate, + ) @workflow_app.command( @@ -457,6 +452,126 @@ def version() -> str: raise typer.Exit() +def _send_grpc(context, parallel_queries, batch, server_url, source): + channel = grpc.insecure_channel(server_url) + stub = garf_pb2_grpc.GarfServiceStub(channel) + rest_context = context.model_dump() + if rest_context.get('writer') == ['console']: + del rest_context['writer'] + else: + rest_context['writer'] = rest_context['writer'][0] + if parallel_queries and len(batch) > 1: + batch = [ + pb.QueryDefinition(title=title, text=text) + for title, text in batch.items() + ] + request = pb.ExecuteBatchRequest( + source=source, batch=batch, context=pb.ExecutionContext(**rest_context) + ) + response = stub.ExecuteBatch(request) + with tracer.start_as_current_span('parse_results batch'): + typer.secho(response.results) + else: + for title, text in batch.items(): + request = pb.ExecuteRequest( + source=source, + title=title, + query=text, + context=pb.ExecutionContext(**rest_context), + ) + response = stub.Execute(request) + with tracer.start_as_current_span('parse_results'): + typer.secho(response.results) + + +def _send_http(context, parallel_queries, batch, server_url, source): + rest_context = context.model_dump() + if rest_context.get('writer') == ['console']: + del rest_context['writer'] + headers = {} + TraceContextTextMapPropagator().inject(headers) + if parallel_queries and len(batch) > 1: + endpoint = f'{server_url}/api/execute:batch' + request = { + 'batch': batch, + 'source': source, + 'context': rest_context, + } + try: + response = requests.post(url=endpoint, json=request, headers=headers) + response.raise_for_status() + except requests.exceptions.HTTPError as e: + raise exceptions.GarfExecutorError( + f'Server error: {e.response.status_code} - {e.response.json()}' + ) from e + + with tracer.start_as_current_span('parse_results batch'): + typer.secho(response.json().get('results')) + else: + endpoint = f'{server_url}/api/execute' + for title, text in batch.items(): + request = { + 'source': source, + 'title': title, + 'query': text, + 'context': rest_context, + } + try: + response = requests.post(url=endpoint, json=request, headers=headers) + response.raise_for_status() + except requests.exceptions.HTTPError as e: + raise exceptions.GarfExecutorError( + f'Server error: {e.response.status_code} - {e.response.json()}' + ) from e + with tracer.start_as_current_span(f'parse_results {title}'): + typer.secho(response.json().get('results')) + + +def _run_worklow_grpc(server_url, workflow_data): + workflow_data.compile() + + channel = grpc.insecure_channel(server_url) + stub = garf_pb2_grpc.GarfServiceStub(channel) + workflow_pb = pb.Workflow() + workflow_dict = workflow_data.model_dump() + config_pb = pb.Config() + if config := workflow_dict.pop('execution_config'): + ParseDict(config, config_pb) + context_pb = pb.ExecutionContext() + if context := workflow_dict.pop('context'): + ParseDict(context, context_pb) + ParseDict(workflow_dict, workflow_pb, ignore_unknown_fields=True) + request = pb.ExecuteWorkflowRequest( + workflow=workflow_pb, + config=config_pb, + context=context_pb, + ) + response = stub.ExecuteWorkflow(request) + with tracer.start_as_current_span('parse_results workflow'): + typer.secho(response.results) + + +def _run_worklow_http(server_url, workflow_file): + headers = {} + TraceContextTextMapPropagator().inject(headers) + endpoint = f'{server_url}/api/execute:workflow' + with open(workflow_file, mode='rb') as f: + files = {'workflow_file': f} + try: + response = requests.post( + url=endpoint, + files=files, + headers=headers, + ) + response.raise_for_status() + except requests.exceptions.HTTPError as e: + raise exceptions.GarfExecutorError( + f'Server error: {e.response.status_code} - {e.response.json()}' + ) from e + with tracer.start_as_current_span('parse_results workflow'): + typer.secho(response.json().get('results')) + + @typer_app.callback( invoke_without_command=True, context_settings={'allow_extra_args': True, 'ignore_unknown_options': True}, diff --git a/libs/executors/garf/executors/garf_pb2.py b/libs/executors/garf/executors/garf_pb2.py index 8ff3703..b1d6092 100644 --- a/libs/executors/garf/executors/garf_pb2.py +++ b/libs/executors/garf/executors/garf_pb2.py @@ -26,7 +26,7 @@ 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\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') +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\"}\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\x12\x17\n\x0fmacro_expansion\x18\x03 \x01(\x08\"\"\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\",\n\x0b\x46\x65tcherInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0f\n\x07version\x18\x02 \x01(\t\"\x85\x01\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\x12,\n\x11required_fetchers\x18\x04 \x03(\x0b\x32\x11.garf.FetcherInfo\"\x9c\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\x12\x1a\n\x12parallel_threshold\x18\x08 \x01(\r\"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\"u\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\x0e\n\x06prefix\x18\x04 \x01(\t\"\xb4\x01\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\x12\x1c\n\x06\x63onfig\x18\x04 \x01(\x0b\x32\x0c.garf.Config\x12\'\n\x07\x63ontext\x18\x05 \x01(\x0b\x32\x16.garf.ExecutionContext\"*\n\x17\x45xecuteWorkflowResponse\x12\x0f\n\x07results\x18\x01 \x03(\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) @@ -44,39 +44,39 @@ _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 + _globals['_QUERYPARAMETERS']._serialized_end=790 + _globals['_EXECUTERESPONSE']._serialized_start=792 + _globals['_EXECUTERESPONSE']._serialized_end=826 + _globals['_QUERYDEFINITION']._serialized_start=828 + _globals['_QUERYDEFINITION']._serialized_end=874 + _globals['_EXECUTEBATCHREQUEST']._serialized_start=876 + _globals['_EXECUTEBATCHREQUEST']._serialized_end=992 + _globals['_EXECUTEBATCHRESPONSE']._serialized_start=994 + _globals['_EXECUTEBATCHRESPONSE']._serialized_end=1033 + _globals['_FETCHERINFO']._serialized_start=1035 + _globals['_FETCHERINFO']._serialized_end=1079 + _globals['_WORKFLOWMETADATA']._serialized_start=1082 + _globals['_WORKFLOWMETADATA']._serialized_end=1215 + _globals['_WORKFLOWSTEP']._serialized_start=1218 + _globals['_WORKFLOWSTEP']._serialized_end=1502 + _globals['_CONFIGMETADATA']._serialized_start=1504 + _globals['_CONFIGMETADATA']._serialized_end=1558 + _globals['_CONFIG']._serialized_start=1561 + _globals['_CONFIG']._serialized_end=1716 + _globals['_WORKFLOW']._serialized_start=1718 + _globals['_WORKFLOW']._serialized_end=1835 + _globals['_EXECUTEWORKFLOWREQUEST']._serialized_start=1838 + _globals['_EXECUTEWORKFLOWREQUEST']._serialized_end=2018 + _globals['_EXECUTEWORKFLOWRESPONSE']._serialized_start=2020 + _globals['_EXECUTEWORKFLOWRESPONSE']._serialized_end=2062 + _globals['_LISTFETCHERSRESPONSE']._serialized_start=2064 + _globals['_LISTFETCHERSRESPONSE']._serialized_end=2122 + _globals['_LISTEXECUTORSRESPONSE']._serialized_start=2124 + _globals['_LISTEXECUTORSRESPONSE']._serialized_end=2164 + _globals['_GARFVERSION']._serialized_start=2166 + _globals['_GARFVERSION']._serialized_end=2196 + _globals['_GARFINFO']._serialized_start=2198 + _globals['_GARFINFO']._serialized_end=2277 + _globals['_GARFSERVICE']._serialized_start=2280 + _globals['_GARFSERVICE']._serialized_end=2812 # @@protoc_insertion_point(module_scope) diff --git a/libs/executors/garf/executors/workflows/workflow.py b/libs/executors/garf/executors/workflows/workflow.py index 07e7f17..28a699e 100644 --- a/libs/executors/garf/executors/workflows/workflow.py +++ b/libs/executors/garf/executors/workflows/workflow.py @@ -20,7 +20,7 @@ import re import urllib from collections import defaultdict -from typing import Any +from typing import Any, Final import pydantic import smart_open @@ -175,6 +175,10 @@ def context(self) -> ExecutionContext: ) +def _ignore_empty_dict(v: Any) -> bool: + return isinstance(v, dict) and not v + + class WorkflowMetadata(pydantic.BaseModel): """Contains optional metadata on workflow. @@ -190,7 +194,7 @@ class WorkflowMetadata(pydantic.BaseModel): version: str | None = None required_garf_version: str | None = None required_fetchers: dict[str, str] | None = pydantic.Field( - default_factory=dict + default_factory=dict, exclude_if=_ignore_empty_dict ) @@ -296,6 +300,8 @@ def compile(self) -> None: else: new_queries.append(query.to_query(self.prefix)) step.queries = new_queries + if self.prefix is not None: + self.prefix = str(self.prefix) @property def attributes(self) -> dict[str, str]: diff --git a/libs/executors/pyproject.toml b/libs/executors/pyproject.toml index b93bd34..a0aa6f0 100644 --- a/libs/executors/pyproject.toml +++ b/libs/executors/pyproject.toml @@ -66,6 +66,7 @@ server=[ "celery[redis]", "opentelemetry-instrumentation-celery", "opentelemetry-instrumentation-redis", + "grpcio-health-checking", ] tests = [ diff --git a/protos/garf.proto b/protos/garf.proto index debe857..cec81a5 100644 --- a/protos/garf.proto +++ b/protos/garf.proto @@ -52,6 +52,7 @@ message ExecutionContext { message QueryParameters { google.protobuf.Struct macro = 1; google.protobuf.Struct template = 2; + bool macro_expansion = 3; } message ExecuteResponse { @@ -74,10 +75,16 @@ message ExecuteBatchResponse { } // Workflow +message FetcherInfo { + string name = 1; + string version = 2; +} + message WorkflowMetadata { string version = 1; string description = 2; string required_garf_version = 3; + repeated FetcherInfo required_fetchers = 4; } message WorkflowStep { @@ -88,6 +95,7 @@ message WorkflowStep { google.protobuf.Struct fetcher_parameters = 5; string writer = 6; google.protobuf.Struct writer_parameters = 7; + uint32 parallel_threshold = 8; } message ConfigMetadata { @@ -106,13 +114,15 @@ message Workflow { string name = 1; repeated WorkflowStep steps = 2; WorkflowMetadata metadata = 3; - Config config = 4; + string prefix = 4; } message ExecuteWorkflowRequest { Workflow workflow = 1; repeated string selected_aliases = 2; repeated string skipped_aliases = 3; + Config config = 4; + ExecutionContext context = 5; } message ExecuteWorkflowResponse { @@ -121,10 +131,6 @@ message ExecuteWorkflowResponse { // List -message FetcherInfo { - string name = 1; - string version = 2; -} message ListFetchersResponse { repeated FetcherInfo results = 1; }