Repository: epoch8/datapipe (language: python)

Issue:
Возможность добавлять мета-информацию в логирование событий
Для эффективной работы с данными из datapipe_events нужна возможность добавлять дополнительные произвольные  поля в вывод

Пример поля event:
```json
{
  "metadata": {
    "flow_name": "some_flow_name",
    "run_uid": "f9e3178a-a44a-4e96-bf85-88517a6631c7",
    "pipeline_id": 5,
  },
  "data": {
    "table_name": "some_table_name",
    "added": 1000,
    "updated": 0,
    "deleted": 0
  }
}
```

Где все поля в `metadata` - специфичны для имплементации t4mp


Discussion / hints:
@elephantum эту задачу уже начал делать Эмиль - @LorinLex
@yumagulovrn а filters тоже в metadata записываются?
> @yumagulovrn а filters тоже в metadata записываются?

@elephantum да, мы будем писать и filters и labels

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4a924bca..f0703a0e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# 0.9.12
+
+* Логгирование меты из `run_config` в `datapipe_events`.
+
 # 0.9.11
 
 * Метадата использует `JSONB` поле в случае Postgres соединения
diff --git a/datapipe/cli.py b/datapipe/cli.py
index 068456a5..5ec3f512 100644
--- a/datapipe/cli.py
+++ b/datapipe/cli.py
@@ -2,9 +2,15 @@
 
 from datapipe.datatable import DataStore
 from datapipe.dsl import Catalog, Pipeline
+from datapipe.step import RunConfig
 
 
-def main(ds: DataStore, catalog: Catalog, pipeline: Pipeline):
+def main(
+    ds: DataStore,
+    catalog: Catalog,
+    pipeline: Pipeline,
+    run_config: RunConfig = None,
+) -> None:
     @click.group()
     @click.option('--debug', is_flag=True, help='Log debug output')
     @click.option('--debug-sql', is_flag=True, help='Log SQL queries VERY VERBOSE')
@@ -21,7 +27,7 @@ def cli(debug: bool, debug_sql: bool) -> None:
     @cli.command()
     def run():
         from .compute import run_pipeline
-        run_pipeline(ds, catalog, pipeline)
+        run_pipeline(ds, catalog, pipeline, run_config)
 
     @cli.command()
     def ui():
diff --git a/datapipe/compute.py b/datapipe/compute.py
index 54bda1e2..6011e415 100644
--- a/datapipe/compute.py
+++ b/datapipe/compute.py
@@ -20,7 +20,7 @@ def run(self, ds: DataStore, run_config: RunConfig = None) -> None:
         gen_process_many(
             self.output_dts,
             self.func,
-            run_config=run_config,
+            run_config=RunConfig.add_labels(run_config, {'step_name': self.name}),
         )
 
 
@@ -36,7 +36,7 @@ def run(self, ds: DataStore, run_config: RunConfig = None) -> None:
             self.output_dts,
             self.func,
             self.chunk_size,
-            run_config=run_config
+            run_config=RunConfig.add_labels(run_config, {'step_name': self.name})
         )
 
 
@@ -109,6 +109,11 @@ def run_steps(ds: DataStore, steps: List[ComputeStep], run_config: RunConfig = N
         step.run(ds, run_config)
 
 
-def run_pipeline(ds: DataStore, catalog: Catalog, pipeline: Pipeline, run_config: RunConfig = None) -> None:
+def run_pipeline(
+    ds: DataStore,
+    catalog: Catalog,
+    pipeline: Pipeline,
+    run_config: RunConfig = None,
+) -> None:
     steps = build_compute(ds, catalog, pipeline)
     run_steps(ds, steps, run_config)
diff --git a/datapipe/datatable.py b/datapipe/datatable.py
index 9dc9bfd3..ceebd7c8 100644
--- a/datapipe/datatable.py
+++ b/datapipe/datatable.py
@@ -45,7 +45,13 @@ def get_metadata(self, idx: Optional[IndexDF] = None) -> MetadataDF:
     def get_data(self, idx: Optional[IndexDF] = None) -> DataDF:
         return self.table_store.read_rows(self.meta_table.get_existing_idx(idx))
 
-    def store_chunk(self, data_df: DataDF, processed_idx: IndexDF = None, now: float = None) -> None:
+    def store_chunk(
+        self,
+        data_df: DataDF,
+        processed_idx: IndexDF = None,
+        now: float = None,
+        run_config: RunConfig = None,
+    ) -> None:
         '''
         Записать новые данные в таблицу.
 
@@ -62,7 +68,10 @@ def store_chunk(self, data_df: DataDF, processed_idx: IndexDF = None, now: float
 
         if len(new_meta_df) > 0 or len(changed_meta_df) > 0:
             self.event_logger.log_state(
-                self.name, added_count=len(new_meta_df), updated_count=len(changed_meta_df), deleted_count=0
+                self.name, added_count=len(new_meta_df),
+                updated_count=len(changed_meta_df),
+                deleted_count=0,
+                run_config=run_config,
             )
 
         # TODO implement transaction meckanism
@@ -78,18 +87,37 @@ def store_chunk(self, data_df: DataDF, processed_idx: IndexDF = None, now: float
             existing_idx = self.meta_table.get_existing_idx(processed_idx)
             deleted_idx = index_difference(existing_idx, data_idx)
 
-            self.table_store.delete_rows(deleted_idx)
-            self.meta_table.mark_rows_deleted(deleted_idx)
+            self.delete_by_idx(deleted_idx, now=now, run_config=run_config)
+
+    def delete_by_idx(
+        self,
+        idx: IndexDF,
+        now: float = None,
+        run_config: RunConfig = None,
+    ) -> None:
+        if len(idx) > 0:
+            logger.debug(f'Deleting {len(idx.index)} rows from {self.name} data')
+            self.event_logger.log_state(
+                self.name,
+                added_count=0,
+                updated_count=0,
+                deleted_count=len(idx),
+                run_config=run_config,
+            )
 
-    def delete_by_idx(self, idx: IndexDF, now: float = None) -> None:
-        self.table_store.delete_rows(idx)
-        self.meta_table.mark_rows_deleted(idx, now=now)
+            self.table_store.delete_rows(idx)
+            self.meta_table.mark_rows_deleted(idx, now=now)
 
-    def delete_stale_by_process_ts(self, process_ts: float, now: float = None, run_config: RunConfig = None) -> None:
+    def delete_stale_by_process_ts(
+        self,
+        process_ts: float,
+        now: float = None,
+        run_config: RunConfig = None,
+    ) -> None:
         for deleted_df in self.meta_table.get_stale_idx(process_ts, run_config=run_config):
             deleted_idx = data_to_index(deleted_df, self.primary_keys)
-            self.table_store.delete_rows(deleted_idx)
-            self.meta_table.mark_rows_deleted(deleted_idx, now=now)
+
+            self.delete_by_idx(deleted_idx, now=now, run_config=run_config)
 
 
 class DataStore:
@@ -113,7 +141,6 @@ def create_table(self, name: str, table_store: TableStore) -> DataTable:
                 dbconn=self.meta_dbconn,
                 name=name,
                 primary_schema=primary_schema,
-                event_logger=self.event_logger,
             ),
             table_store=table_store,
             event_logger=self.event_logger,
@@ -277,11 +304,11 @@ def gen_process_many(
 
             # TODO перенести get_process* в compute.BatchGenerateStep и пользоваться event_logger из metastore
             if dts:
-                dts[0].event_logger.log_exception(e)
+                dts[0].event_logger.log_exception(e, run_config=run_config)
             return
 
         for k, dt_k in enumerate(dts):
-            dt_k.store_chunk(chunk_dfs[k])
+            dt_k.store_chunk(chunk_dfs[k], run_config=run_config)
 
     for k, dt_k in enumerate(dts):
         dt_k.delete_stale_by_process_ts(now, run_config=run_config)
@@ -360,7 +387,7 @@ def inc_process_many(
                     chunks_df = proc_func(*input_dfs, **kwargs)
                 except Exception as e:
                     logger.error(f"Transform failed ({proc_func.__name__}): {str(e)}")
-                    ds.event_logger.log_exception(e)
+                    ds.event_logger.log_exception(e, run_config=run_config)
 
                     continue
 
@@ -369,11 +396,15 @@ def inc_process_many(
                     chunk_df_k = chunks_df[k] if len(res_dts) > 1 else chunks_df
 
                     # Добавляем результат в результирующие чанки
-                    res_dt.store_chunk(data_df=chunk_df_k, processed_idx=idx)
+                    res_dt.store_chunk(
+                        data_df=chunk_df_k,
+                        processed_idx=idx,
+                        run_config=run_config,
+                    )
 
             else:
                 for k, res_dt in enumerate(res_dts):
-                    res_dt.delete_by_idx(idx)
+                    res_dt.delete_by_idx(idx, run_config=run_config)
 
 
 class ExternalTableUpdater(ComputeStep):
@@ -392,7 +423,11 @@ def run(self, ds: DataStore, run_config: RunConfig = None) -> None:
 
             if len(new_meta_df) > 0 or len(changed_meta_df) > 0:
                 ds.event_logger.log_state(
-                    self.name, added_count=len(new_meta_df), updated_count=len(changed_meta_df), deleted_count=0
+                    self.name,
+                    added_count=len(new_meta_df),
+                    updated_count=len(changed_meta_df),
+                    deleted_count=0,
+                    run_config=run_config,
                 )
 
             # TODO switch to iterative store_chunk and self.table.sync_meta_by_process_ts
@@ -401,4 +436,13 @@ def run(self, ds: DataStore, run_config: RunConfig = None) -> None:
             self.table.meta_table.update_meta_for_store_chunk(changed_meta_df)
 
         for stale_idx in self.table.meta_table.get_stale_idx(now):
+            logger.debug(f'Deleting {len(stale_idx.index)} rows from {self.name} data')
+            self.output_dts[0].event_logger.log_state(
+                self.name,
+                added_count=0,
+                updated_count=0,
+                deleted_count=len(stale_idx),
+                run_config=run_config,
+            )
+
             self.table.meta_table.mark_rows_deleted(stale_idx, now=now)
diff --git a/datapipe/event_logger.py b/datapipe/event_logger.py
index 90326df4..b92a6b2b 100644
--- a/datapipe/event_logger.py
+++ b/datapipe/event_logger.py
@@ -9,6 +9,8 @@
 from sqlalchemy.sql.sqltypes import DateTime, Integer, String, JSON
 from sqlalchemy.dialects.postgresql import JSONB
 
+from datapipe.step import RunConfig
+
 
 logger = logging.getLogger('datapipe.event_logger')
 
@@ -41,39 +43,81 @@ def _make_table_schema(self, dbconn: 'DBConn'):
             Column('event', JSON if dbconn.con.name == 'sqlite' else JSONB)
         ]
 
-    def log_state(self, table_name, added_count, updated_count, deleted_count):
+    def log_state(
+        self,
+        table_name,
+        added_count,
+        updated_count,
+        deleted_count,
+        run_config: RunConfig = None,
+    ):
         logger.debug(f'Table "{table_name}": added = {added_count}; updated = {updated_count}; deleted = {deleted_count}')
 
+        if run_config is not None:
+            meta = {
+                "labels": run_config.labels,
+                "filters": run_config.filters,
+            }
+        else:
+            meta = {}
+
         ins = self.events_table.insert().values(
             type=EventTypes.STATE.value,
             event={
-                "table_name": table_name,
-                "added_count": added_count,
-                "updated_count": updated_count,
-                "deleted_count": deleted_count
+                "meta": meta,
+                "data": {
+                    "table_name": table_name,
+                    "added_count": added_count,
+                    "updated_count": updated_count,
+                    "deleted_count": deleted_count,
+                }
             }
         )
 
         self.dbconn.con.execute(ins)
 
-    def log_error(self, type, message, description, params):
-        logger.debug(f'Error {type} {message}')
+    def log_error(
+        self,
+        type,
+        message,
+        description,
+        params,
+        run_config: RunConfig = None,
+    ) -> None:
+        if run_config is not None:
+            logger.debug(f'Error in step {run_config.labels.get("step_name")}: {type} {message}')
+            meta = {
+                "labels": run_config.labels,
+                "filters": run_config.filters,
+            }
+        else:
+            logger.debug(f'Error: {type} {message}')
+            meta = {}
+
         ins = self.events_table.insert().values(
             type=EventTypes.ERROR.value,
             event={
-                "type": type,
-                "message": message,
-                "description": description,
-                "params": params
+                "meta": meta,
+                "data": {
+                    "type": type,
+                    "message": message,
+                    "description": description,
+                    "params": params,
+                }
             }
         )
 
         self.dbconn.con.execute(ins)
 
-    def log_exception(self, exc: Exception):
+    def log_exception(
+        self,
+        exc: Exception,
+        run_config: RunConfig = None,
+    ) -> None:
         self.log_error(
             type=type(exc).__name__,
             message=str(exc),
             description=traceback.format_exc(),
-            params=exc.args
+            params=exc.args,
+            run_config=run_config,
         )
diff --git a/datapipe/metastore.py b/datapipe/metastore.py
index a680790e..f26bdbc7 100644
--- a/datapipe/metastore.py
+++ b/datapipe/metastore.py
@@ -13,7 +13,6 @@
 
 from datapipe.types import IndexDF, DataSchema, DataDF, MetadataDF, data_to_index
 from datapipe.store.database import DBConn, sql_apply_runconfig_filter, sql_schema_to_sqltype
-from datapipe.event_logger import EventLogger
 from datapipe.step import RunConfig
 
 
@@ -40,16 +39,10 @@ def __init__(
         dbconn: DBConn,
         name: str,
         primary_schema: DataSchema,
-        event_logger: EventLogger = None
     ):
         self.dbconn = dbconn
         self.name = name
 
-        if event_logger is None:
-            self.event_logger = EventLogger(dbconn)
-        else:
-            self.event_logger = event_logger
-
         self.primary_keys = [column.name for column in primary_schema]
 
         for item in primary_schema:
@@ -288,11 +281,12 @@ def update_meta_for_store_chunk(self, changed_meta_df: MetadataDF) -> None:
         if len(changed_meta_df) > 0:
             self._update_existing_metadata_rows(changed_meta_df)
 
-    def mark_rows_deleted(self, deleted_idx: IndexDF, now: float = None) -> None:
+    def mark_rows_deleted(
+        self,
+        deleted_idx: IndexDF,
+        now: float = None,
+    ) -> None:
         if len(deleted_idx) > 0:
-            logger.debug(f'Deleting {len(deleted_idx.index)} rows from {self.name} data')
-            self.event_logger.log_state(self.name, added_count=0, updated_count=0, deleted_count=len(deleted_idx))
-
             if now is None:
                 now = time.time()
 
diff --git a/datapipe/step.py b/datapipe/step.py
index 7cea345a..3fae371b 100644
--- a/datapipe/step.py
+++ b/datapipe/step.py
@@ -1,17 +1,31 @@
-from typing import List, TYPE_CHECKING, Dict, Any
-from dataclasses import dataclass
+from typing import List, TYPE_CHECKING, Dict, Any, Optional
+from dataclasses import dataclass, field
 
 if TYPE_CHECKING:
     from datapipe.datatable import DataStore, DataTable
 
 
+LabelDict = Dict[str, Any]
+
+
 @dataclass
 class RunConfig:
     # Массив глобально применяемых фильтров
     # если не пуст, то во время запуска обрабатываются только те строки,
     # которые строго соответствуют фильтру
     # (в случае, если у таблицы есть идентификатор с совпадающим именем).
-    filters: Dict[str, Any]
+    filters: LabelDict = field(default_factory=dict)
+    labels: LabelDict = field(default_factory=dict)
+
+    @classmethod
+    def add_labels(cls, rc: Optional['RunConfig'], labels: LabelDict) -> 'RunConfig':
+        if rc is not None:
+            return RunConfig(
+                filters=rc.filters,
+                labels={**rc.labels, **labels},
+            )
+        else:
+            return RunConfig(labels=labels)
 
 
 @dataclass
diff --git a/examples/v0.9.0/one_to_many_pipeline.py b/examples/v0.9.0/one_to_many_pipeline.py
index eeb217f3..b2a0aae6 100644
--- a/examples/v0.9.0/one_to_many_pipeline.py
+++ b/examples/v0.9.0/one_to_many_pipeline.py
@@ -8,11 +8,20 @@
 from datapipe.dsl import Catalog, Pipeline, Table, BatchGenerate, BatchTransform
 from datapipe.store.database import DBConn
 from datapipe.cli import main
+from datapipe.step import RunConfig
 
 
 dbconn = DBConn('sqlite:///db.sqlite')
 ds = DataStore(dbconn)
 
+run_config = RunConfig(
+    filters={},
+    labels={
+        "pipeline_name": 'test_name',
+        "pipeline_id": 1
+    }
+)
+
 
 def generate_products():
     for pipeline_id in range(1, 3):
@@ -242,4 +251,4 @@ def filter(df: pd.DataFrame) -> pd.DataFrame:
 ])
 
 
-main(ds, catalog, pipeline)
+main(ds, catalog, pipeline, run_config)
diff --git a/pyproject.toml b/pyproject.toml
index 00966fe0..df05a01a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "datapipe"
-version = "0.9.11"
+version = "0.9.12"
 description = ""
 authors = ["Andrey Tatarinov <a@tatarinov.co>"]
 


Test patch:
diff --git a/tests/test_datatable.py b/tests/test_datatable.py
index 4f865217..4a86a583 100644
--- a/tests/test_datatable.py
+++ b/tests/test_datatable.py
@@ -127,14 +127,14 @@ def gen():
     def func():
         return TEST_DF
 
-    gen_process(
-        tbl1_gen,
+    gen_process_many(
+        [tbl1_gen],
         gen
     )
 
     with pytest.raises(Exception):
-        gen_process(
-            tbl1,
+        gen_process_many(
+            [tbl1],
             func
         )
 
@@ -152,8 +152,8 @@ def func2():
     )
 
     with pytest.raises(Exception):
-        gen_process(
-            tbl1,
+        gen_process_many(
+            [tbl1],
             func2
         )
 
diff --git a/tests/test_meta_info_in_datapipe_events.py b/tests/test_meta_info_in_datapipe_events.py
new file mode 100644
index 00000000..fd7c7ecc
--- /dev/null
+++ b/tests/test_meta_info_in_datapipe_events.py
@@ -0,0 +1,119 @@
+import pandas as pd
+from sqlalchemy.sql.expression import select
+
+from datapipe.step import RunConfig
+from datapipe.store.database import TableStoreDB
+from datapipe.datatable import DataStore
+from datapipe.dsl import Catalog, Pipeline, BatchTransform, BatchGenerate,\
+    Table
+from datapipe.compute import run_pipeline
+
+from sqlalchemy.sql.schema import Column
+from sqlalchemy.sql.sqltypes import Integer, JSON
+
+
+TEST_SCHEMA = [
+    Column('pipeline_id', Integer(), primary_key=True),
+    Column('offer_id', Integer(), primary_key=True),
+    Column('test_field', JSON)
+]
+
+
+def generate_data():
+    df_data = [{
+        "pipeline_id": 1,
+        "offer_id": 1,
+        "test_field": {"a": 1}
+    }]
+    yield pd.DataFrame(data=df_data)
+
+
+def update_data(df: pd.DataFrame) -> pd.DataFrame:
+    df["test_field"].apply(lambda x: {**x, "b": 2})
+    df.index = df.index.astype('str')
+    return df
+
+
+def test_meta_info_in_datapipe_events(dbconn) -> None:
+    ds = DataStore(dbconn)
+
+    run_config = RunConfig(
+        filters={
+            "pipeline_id": 1
+        },
+        labels={
+            "pipeline_name": 'test_name',
+            "pipeline_id": 1
+        }
+    )
+
+    catalog = Catalog({
+        'test_generate': Table(
+            store=TableStoreDB(
+                dbconn,
+                'test_generate_data',
+                TEST_SCHEMA
+            )
+        ),
+        'test_transform': Table(
+            store=TableStoreDB(
+                dbconn,
+                'test_transform_data',
+                TEST_SCHEMA
+            )
+        )
+    })
+
+    pipeline = Pipeline([
+        BatchGenerate(
+            generate_data,
+            outputs=["test_generate"],
+        ),
+        BatchTransform(
+            update_data,
+            inputs=["test_generate"],
+            outputs=["test_transform"],
+        )
+    ])
+
+    run_pipeline(ds, catalog, pipeline, run_config)
+
+    df_events = pd.read_sql_query(select(catalog.get_datatable(ds, 'test_generate').event_logger.events_table), dbconn.con)
+
+    assert df_events.loc[0]["event"] == {
+        "meta": {
+            "labels": {
+                "step_name": "generate_data",
+                "pipeline_name": "test_name",
+                "pipeline_id": 1,
+            },
+            "filters": {
+                "pipeline_id": 1,
+            }
+        },
+        "data": {
+            "table_name": "test_generate",
+            "added_count": 1,
+            "updated_count": 0,
+            "deleted_count": 0
+        }
+    }
+
+    assert df_events.loc[1]["event"] == {
+        "meta": {
+            "labels": {
+                "step_name": "update_data",
+                "pipeline_name": "test_name",
+                "pipeline_id": 1,
+            },
+            "filters": {
+                "pipeline_id": 1,
+            }
+        },
+        "data": {
+            "table_name": "test_transform",
+            "added_count": 1,
+            "updated_count": 0,
+            "deleted_count": 0,
+        }
+    }