TECH-61: progressor backend implementation (#25)

Dockerfile.neighbour

@@ -0,0 +1,46 @@
+ARG OTP_VERSION
+
+# Build the release
+FROM docker.io/library/erlang:${OTP_VERSION} AS builder
+SHELL ["/bin/bash", "-o", "pipefail", "-c"]
+
+# Install thrift compiler
+ARG THRIFT_VERSION
+ARG TARGETARCH
+
+RUN wget -q -O- "https://github.com/valitydev/thrift/releases/download/${THRIFT_VERSION}/thrift-${THRIFT_VERSION}-linux-${TARGETARCH}.tar.gz" \
+    | tar -xvz -C /usr/local/bin/
+
+# Copy sources
+RUN mkdir /build
+COPY . /build/
+
+# Build the release
+WORKDIR /build
+RUN rebar3 compile \
+    && rebar3 as test_neighbour release
+
+# Make a runner image
+FROM docker.io/library/erlang:${OTP_VERSION}-slim
+
+ARG SERVICE_NAME=mg_cth_neighbour
+ARG USER_UID=1001
+ARG USER_GID=$USER_UID
+
+# Set env
+ENV CHARSET=UTF-8
+ENV LANG=C.UTF-8
+
+# Expose SERVICE_NAME as env so CMD expands properly on start
+ENV SERVICE_NAME=${SERVICE_NAME}
+
+# Set runtime
+WORKDIR /opt/${SERVICE_NAME}
+
+COPY --from=builder /build/_build/test_neighbour/rel/${SERVICE_NAME} /opt/${SERVICE_NAME}
+
+RUN echo "#!/bin/sh" >> /entrypoint.sh && \
+    echo "exec /opt/${SERVICE_NAME}/bin/${SERVICE_NAME} foreground" >> /entrypoint.sh && \
+    chmod +x /entrypoint.sh
+
+ENTRYPOINT ["/entrypoint.sh"]

Makefile

@@ -14,15 +14,19 @@ DOTENV := $(shell grep -v '^\#' .env)
 DEV_IMAGE_TAG = $(TEST_CONTAINER_NAME)-dev
 DEV_IMAGE_ID = $(file < .image.dev)
 
+TEST_IMAGE_TAG = $(DIST_CONTAINER_NAME)-test
+TEST_IMAGE_ID = $(file < .image.test)
+
 DOCKER ?= docker
 DOCKERCOMPOSE ?= docker-compose
 DOCKERCOMPOSE_W_ENV = DEV_IMAGE_TAG=$(DEV_IMAGE_TAG) $(DOCKERCOMPOSE) -f compose.yaml -f compose.tracing.yaml
 REBAR ?= rebar3
 TEST_CONTAINER_NAME ?= testrunner
+DIST_CONTAINER_NAME ?= distrunner
 
 all: compile xref lint check-format dialyze eunit
 
-.PHONY: dev-image clean-dev-image wc-shell test
+.PHONY: dev-image test-image clean-dev-image clean-test-image wc-shell test
 
 dev-image: .image.dev
 
@@ -36,6 +40,18 @@ ifneq ($(DEV_IMAGE_ID),)
 	rm .image.dev
 endif
 
+test-image: .image.test
+
+.image.test: Dockerfile.neighbour .env
+	env $(DOTENV) $(DOCKERCOMPOSE_W_ENV) build $(DIST_CONTAINER_NAME)
+	$(DOCKER) image ls -q -f "reference=$(TEST_IMAGE_ID)" | head -n1 > $@
+
+clean-test-image:
+ifneq ($(TEST_IMAGE_ID),)
+	$(DOCKER) image rm -f $(TEST_IMAGE_TAG)
+	rm .image.test
+endif
+
 DOCKER_WC_OPTIONS := -v $(PWD):$(PWD) --workdir $(PWD)
 DOCKER_WC_EXTRA_OPTIONS ?= --rm
 DOCKER_RUN = $(DOCKER) run -t $(DOCKER_WC_OPTIONS) $(DOCKER_WC_EXTRA_OPTIONS)

apps/machinegun/src/machinegun.app.src

@@ -26,6 +26,8 @@
         erl_health,
         prometheus,
         prometheus_cowboy,
+        progressor,
+        mg_progressor,
         mg_utils,
         mg_core,
         mg_riak,

apps/mg_conf/src/mg_conf.erl

@@ -99,7 +99,8 @@ events_machine_options(NS, NSs, Pulse) ->
         event_sinks => EventSinks,
         pulse => Pulse,
         default_processing_timeout => maps:get(default_processing_timeout, NSConfigs),
-        event_stash_size => maps:get(event_stash_size, NSConfigs, 0)
+        event_stash_size => maps:get(event_stash_size, NSConfigs, 0),
+        engine => maps:get(engine, NSConfigs, machinegun)
     }.
 
 -spec machine_options(mg_core:ns(), events_machines(), pulse()) -> mg_core_machine:options().

apps/mg_core/src/mg_core_events_machine.erl

@@ -127,7 +127,8 @@
     pulse => mpulse:handler(),
     event_sinks => [mg_core_event_sink:handler()],
     default_processing_timeout => timeout(),
-    event_stash_size => non_neg_integer()
+    event_stash_size => non_neg_integer(),
+    engine => machinegun | progressor
 }.
 % like mg_core_storage:options() except `name`
 -type storage_options() :: mg_utils:mod_opts(map()).

apps/mg_cth/src/mg_cth.erl

@@ -69,9 +69,48 @@ kafka_client_config(Brokers) ->
         ]}
     ].
 
+-spec epg_connector_config() -> _.
+epg_connector_config() ->
+    [
+        {databases, #{
+            progressor_db => #{
+                host => "postgres",
+                port => 5432,
+                database => "progressor_db",
+                username => "progressor",
+                password => "progressor"
+            }
+        }},
+        {pools, #{
+            default_pool => #{
+                database => progressor_db,
+                size => 10
+            }
+        }}
+    ].
+
+-spec progressor_config() -> _.
+progressor_config() ->
+    [
+        {namespaces, #{
+            binary_to_atom(?NS) => #{
+                processor => #{client => null},
+                storage => #{
+                    client => prg_pg_backend,
+                    options => #{pool => default_pool}
+                },
+                worker_pool_size => 0
+            }
+        }}
+    ].
+
 -spec start_application(app()) -> _Deps :: [appname()].
 start_application(brod) ->
     genlib_app:start_application_with(brod, kafka_client_config(?BROKERS));
+start_application(epg_connector) ->
+    genlib_app:start_application_with(epg_connector, epg_connector_config());
+start_application(progressor) ->
+    genlib_app:start_application_with(progressor, progressor_config());
 start_application({AppName, Env}) ->
     genlib_app:start_application_with(AppName, Env);
 start_application(AppName) ->

apps/mg_cth_neighbour/README.md

@@ -0,0 +1,4 @@
+mg_cth_neighbour
+=====
+
+Вспомогательное приложения для тестов, в релиз не включается.

apps/mg_cth_neighbour/rebar.config

@@ -0,0 +1,7 @@
+{erl_opts, [debug_info]}.
+{deps, []}.
+
+{shell, [
+    %% {config, "config/sys.config"},
+    {apps, [mg_cth_neighbour]}
+]}.

apps/mg_cth_neighbour/src/mg_cth_nbr_processor.erl

@@ -0,0 +1,59 @@
+-module(mg_cth_nbr_processor).
+
+-export([process/3]).
+
+-spec process({_TaskType, _Args, _Process}, _Opts, _Ctx) -> _.
+process({init, _InitArgs, _Process}, _Opts, _Ctx) ->
+    Result = #{events => [event(1)]},
+    {ok, Result};
+process(
+    %% <<"simple_call">>
+    {call, <<131, 109, 0, 0, 0, 11, 115, 105, 109, 112, 108, 101, 95, 99, 97, 108, 108>> = CallArgs, _Process},
+    _Opts,
+    _Ctx
+) ->
+    Result = #{
+        response => CallArgs,
+        events => []
+    },
+    {ok, Result};
+process(
+    %% <<"fail_call">>
+    {call, <<131, 109, 0, 0, 0, 9, 102, 97, 105, 108, 95, 99, 97, 108, 108>> = _CallArgs, _Process},
+    _Opts,
+    _Ctx
+) ->
+    {error, do_not_retry};
+process(
+    %% simple repair
+    {timeout, _Args, _Process},
+    _Opts,
+    _Ctx
+) ->
+    Result = #{events => []},
+    {ok, Result};
+%
+process(
+    %% <<"repair_ok">>
+    {repair, <<131, 109, 0, 0, 0, 9, 114, 101, 112, 97, 105, 114, 95, 111, 107>> = CallArgs, _Process},
+    _Opts,
+    _Ctx
+) ->
+    Result = #{events => [], response => CallArgs},
+    {ok, Result};
+process(
+    %% <<"repair_fail">>
+    {repair, <<131, 109, 0, 0, 0, 11, 114, 101, 112, 97, 105, 114, 95, 102, 97, 105, 108>> = _CallArgs, _Process},
+    _Opts,
+    _Ctx
+) ->
+    {error, unreach}.
+
+%
+event(Id) ->
+    #{
+        event_id => Id,
+        timestamp => erlang:system_time(second),
+        metadata => #{<<"format_version">> => 1},
+        payload => erlang:term_to_binary({bin, crypto:strong_rand_bytes(8)})
+    }.

apps/mg_cth_neighbour/src/mg_cth_neighbour.app.src

@@ -0,0 +1,15 @@
+{application, mg_cth_neighbour, [
+    {description, "An OTP application"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {mod, {mg_cth_neighbour_app, []}},
+    {applications, [
+        kernel,
+        stdlib,
+        progressor
+    ]},
+    {env, []},
+    {modules, []},
+    {licenses, ["Apache-2.0"]},
+    {links, []}
+ ]}.

apps/mg_cth_neighbour/src/mg_cth_neighbour_app.erl

@@ -0,0 +1,20 @@
+%%%-------------------------------------------------------------------
+%% @doc mg_cth_neighbour public API
+%% @end
+%%%-------------------------------------------------------------------
+
+-module(mg_cth_neighbour_app).
+
+-behaviour(application).
+
+-export([start/2, stop/1]).
+
+-spec start(_, _) -> _.
+start(_StartType, _StartArgs) ->
+    mg_cth_neighbour_sup:start_link().
+
+-spec stop(_) -> _.
+stop(_State) ->
+    ok.
+
+%% internal functions

apps/mg_cth_neighbour/src/mg_cth_neighbour_sup.erl

@@ -0,0 +1,39 @@
+%%%-------------------------------------------------------------------
+%% @doc mg_cth_neighbour top level supervisor.
+%% @end
+%%%-------------------------------------------------------------------
+
+-module(mg_cth_neighbour_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0]).
+
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+-spec start_link() -> _.
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+%% sup_flags() = #{strategy => strategy(),         % optional
+%%                 intensity => non_neg_integer(), % optional
+%%                 period => pos_integer()}        % optional
+%% child_spec() = #{id => child_id(),       % mandatory
+%%                  start => mfargs(),      % mandatory
+%%                  restart => restart(),   % optional
+%%                  shutdown => shutdown(), % optional
+%%                  type => worker(),       % optional
+%%                  modules => modules()}   % optional
+-spec init(_) -> _.
+init([]) ->
+    SupFlags = #{
+        strategy => one_for_all,
+        intensity => 0,
+        period => 1
+    },
+    ChildSpecs = [],
+    {ok, {SupFlags, ChildSpecs}}.
+
+%% internal functions

apps/mg_progressor/rebar.config

@@ -0,0 +1,2 @@
+{erl_opts, [debug_info]}.
+{deps, []}.

apps/mg_progressor/src/mg_progressor.app.src

@@ -0,0 +1,13 @@
+{application, mg_progressor, [
+    {description, "An OTP library"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {applications, [
+        kernel,
+        stdlib
+    ]},
+    {env, []},
+    {modules, []},
+    {licenses, []},
+    {links, []}
+ ]}.