Playground: Basic server implementation

[vadimalekseev]

Fixes #403

Examples:

Debug actions with given config

It sends two logs:

1. {"message": "file.d playground"}
2. {"hello": "playground"}

With given config:

actions:
- type: set_time
  field: time
  metric_name: input
- type: rename
  time: ts
- type: modify
  hello: world

And returns the result of processing, debug information and metrics

$ curl -X POST 'http://127.0.0.1:5950/api/v1/do-actions' -d '{
  "events": [{"pipeline_kafka_topic":"obs-seq-db-logs","time":"2023-08-22T10:01:09.282965716Z","stream":"stdout","k8s_node":"infrakuben65329z501","k8s_namespace":"logging","k8s_pod":"seq-proxy-prod-z501-6ccc888994-nkj94","k8s_container":"seq-proxy","k8s_pod_label_app":"seq-proxy-prod-z501","zone":"z501","k8s_cluster":"obs","env":"infra-ts","level":"informational","ts":1692698469,"message":"search","req":{"query":"(access_token_leaked: \"jwt_token\" AND service: \"adms-auth\")","offset":0,"size":2,"from":"2023-08-21 10:01:09","to":"2023-08-22 10:01:09","with_total":false,"explain":false,"agg_name":"","agg_field":"","agg_filter":"","interval":"0s"}}],
  "actions": [
    {"type": "modify", "from":"${req.from}", "to":"${req.to}"},
      {
        "type": "convert_date",
        "field": "from",
        "source_formats": ["rfc3339nano", "rfc3339", "2006-01-02 15:04:05"],
        "target_format": "rfc3339nano",
        "remove_on_fail": false
      },
      {
        "type": "convert_date",
        "field": "to",
        "source_formats": ["rfc3339nano", "rfc3339", "2006-01-02 15:04:05"],
        "target_format": "rfc3339nano",
        "remove_on_fail": false
      }
  ],
  "debug": false
}'

# stdout
{
  "result": [
    {
      "event": {
        "message": "file.d playground",
        "ts": "2023-07-02T19:48:20.971557+03:00",
        "hello": "world"
      }
    },
    {
      "event": {
        "hello": "world",
        "ts": "2023-07-02T19:48:20.971674+03:00"
      }
    }
  ],
  "stdout": "{\"level\":\"info\",\"ts\":\"00.000276\",\"msg\":\"starting pipeline \\\"test\\\": procs=8\"}\n{\"level\":\"info\",\"ts\":\"00.000357\",\"msg\":\"starting output plugin \\\"devnull\\\"\"}\n{\"level\":\"info\",\"ts\":\"00.000359\",\"msg\":\"stating processors, count=8\"}\n{\"level\":\"info\",\"ts\":\"00.001165\",\"msg\":\"starting input plugin \\\"fake\\\"\"}\n{\"level\":\"info\",\"ts\":\"00.001753\",\"logger\":\"action.debug\",\"msg\":\"initial\",\"offset\":0,\"event\":{\"message\":\"file.d playground\"}}\n{\"level\":\"info\",\"ts\":\"00.001819\",\"logger\":\"action.debug\",\"msg\":\"set_time\",\"offset\":0,\"event\":{\"message\":\"file.d playground\",\"time\":\"2023-07-02T19:48:20.971557+03:00\"}}\n{\"level\":\"info\",\"ts\":\"00.001828\",\"logger\":\"action.debug\",\"msg\":\"rename\",\"offset\":0,\"event\":{\"message\":\"file.d playground\",\"ts\":\"2023-07-02T19:48:20.971557+03:00\"}}\n{\"level\":\"info\",\"ts\":\"00.001839\",\"logger\":\"action.debug\",\"msg\":\"modify\",\"offset\":0,\"event\":{\"message\":\"file.d playground\",\"ts\":\"2023-07-02T19:48:20.971557+03:00\",\"hello\":\"world\"}}\n{\"level\":\"info\",\"ts\":\"00.001858\",\"logger\":\"action.debug\",\"msg\":\"initial\",\"offset\":1,\"event\":{\"hello\":\"playground\"}}\n{\"level\":\"info\",\"ts\":\"00.001931\",\"logger\":\"action.debug\",\"msg\":\"set_time\",\"offset\":1,\"event\":{\"hello\":\"playground\",\"time\":\"2023-07-02T19:48:20.971674+03:00\"}}\n{\"level\":\"info\",\"ts\":\"00.001937\",\"logger\":\"action.debug\",\"msg\":\"rename\",\"offset\":1,\"event\":{\"hello\":\"playground\",\"ts\":\"2023-07-02T19:48:20.971674+03:00\"}}\n{\"level\":\"info\",\"ts\":\"00.001941\",\"logger\":\"action.debug\",\"msg\":\"modify\",\"offset\":1,\"event\":{\"hello\":\"world\",\"ts\":\"2023-07-02T19:48:20.971674+03:00\"}}\n{\"level\":\"info\",\"ts\":\"00.001955\",\"msg\":\"stopping pipeline \\\"test\\\", total committed=2\"}\n{\"level\":\"info\",\"ts\":\"00.001957\",\"msg\":\"stopping processors count=8\"}\n{\"level\":\"info\",\"ts\":\"00.001962\",\"msg\":\"stopping \\\"test\\\" input\"}\n{\"level\":\"info\",\"ts\":\"00.001969\",\"msg\":\"stopping \\\"test\\\" output\"}\n",
  "metrics": "# HELP file_d_pipeline_test_antispam_active Gauge indicates whether the antispam is enabled\n# TYPE file_d_pipeline_test_antispam_active gauge\nfile_d_pipeline_test_antispam_active 0.0\n\n# HELP file_d_pipeline_test_event_pool_capacity Pool capacity value\n# TYPE file_d_pipeline_test_event_pool_capacity gauge\nfile_d_pipeline_test_event_pool_capacity 1.0\n\n# HELP file_d_pipeline_test_input_events_count how many events processed by pipeline \\\"test\\\" and #1 action\n# TYPE file_d_pipeline_test_input_events_count counter\nfile_d_pipeline_test_input_events_count_total{gen=\"0\",status=\"passed\",version=\"\"} 2.0\nfile_d_pipeline_test_input_events_count_total{gen=\"0\",status=\"received\",version=\"\"} 2.0\n\n# HELP file_d_pipeline_test_input_events_size total size of events processed by pipeline \\\"test\\\" and #1 action\n# TYPE file_d_pipeline_test_input_events_size counter\nfile_d_pipeline_test_input_events_size_total{gen=\"0\",status=\"passed\",version=\"\"} 55.0\nfile_d_pipeline_test_input_events_size_total{gen=\"0\",status=\"received\",version=\"\"} 55.0\n\n"
}

Unescaped stdout from the response:

{"level":"info","ts":"00.000663","msg":"starting pipeline \"test\": procs=8"}
{"level":"info","ts":"00.000890","msg":"starting output plugin \"devnull\""}
{"level":"info","ts":"00.000899","msg":"stating processors, count=8"}
{"level":"info","ts":"00.001037","msg":"starting input plugin \"fake\""}
{"level":"info","ts":"00.001110","logger":"action.debug","msg":"initial","offset":0,"event":{"message":"file.d playground"}}
{"level":"info","ts":"00.001210","logger":"action.debug","msg":"set_time","offset":0,"event":{"message":"file.d playground","time":"2023-07-02T19:54:45.5609+03:00"}}
{"level":"info","ts":"00.001225","logger":"action.debug","msg":"rename","offset":0,"event":{"message":"file.d playground","ts":"2023-07-02T19:54:45.5609+03:00"}}
{"level":"info","ts":"00.001266","logger":"action.debug","msg":"modify","offset":0,"event":{"message":"file.d playground","ts":"2023-07-02T19:54:45.5609+03:00","hello":"world"}}
{"level":"info","ts":"00.001340","logger":"action.debug","msg":"initial","offset":1,"event":{"hello":"playground"}}
{"level":"info","ts":"00.001464","logger":"action.debug","msg":"set_time","offset":1,"event":{"hello":"playground","time":"2023-07-02T19:54:45.561191+03:00"}}
{"level":"info","ts":"00.001472","logger":"action.debug","msg":"rename","offset":1,"event":{"hello":"playground","ts":"2023-07-02T19:54:45.561191+03:00"}}
{"level":"info","ts":"00.001479","logger":"action.debug","msg":"modify","offset":1,"event":{"hello":"world","ts":"2023-07-02T19:54:45.561191+03:00"}}
{"level":"info","ts":"00.001494","msg":"stopping pipeline \"test\", total committed=2"}
{"level":"info","ts":"00.001497","msg":"stopping processors count=8"}
{"level":"info","ts":"00.001506","msg":"stopping \"test\" input"}
{"level":"info","ts":"00.001509","msg":"stopping \"test\" output"}

Unescaped metrics from the response:

# HELP file_d_pipeline_test_antispam_active Gauge indicates whether the antispam is enabled
# TYPE file_d_pipeline_test_antispam_active gauge
file_d_pipeline_test_antispam_active 0.0

# HELP file_d_pipeline_test_event_pool_capacity Pool capacity value
# TYPE file_d_pipeline_test_event_pool_capacity gauge
file_d_pipeline_test_event_pool_capacity 1.0

# HELP file_d_pipeline_test_input_events_count how many events processed by pipeline \"test\" and #1 action
# TYPE file_d_pipeline_test_input_events_count counter
file_d_pipeline_test_input_events_count_total{gen="0",status="passed",version=""} 2.0
file_d_pipeline_test_input_events_count_total{gen="0",status="received",version=""} 2.0

# HELP file_d_pipeline_test_input_events_size total size of events processed by pipeline \"test\" and #1 action
# TYPE file_d_pipeline_test_input_events_size counter
file_d_pipeline_test_input_events_size_total{gen="0",status="passed",version=""} 55.0
file_d_pipeline_test_input_events_size_total{gen="0",status="received",version=""} 55.0

[codecov[bot]]

Codecov Report

Attention: 30 lines in your changes are missing coverage. Please review.

Comparison is base (1b30680) 67.82% compared to head (38bd596) 67.99%. Report is 1 commits behind head on master.

:exclamation: Current head 38bd596 differs from pull request most recent head a3d9080. Consider uploading reports for the commit a3d9080 to get more accurate results

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #404      +/-   ##
==========================================
+ Coverage   67.82%   67.99%   +0.17%     
==========================================
  Files         104      105       +1     
  Lines       10690    10512     -178     
==========================================
- Hits         7250     7148     -102     
+ Misses       2977     2892      -85     
- Partials      463      472       +9     

Files	Coverage Δ
pipeline/event.go	83.33% <100.00%> (-2.31%)	:arrow_down:
pipeline/pipeline.go	66.85% <100.00%> (+5.73%)	:arrow_up:
test/test.go	92.85% <100.00%> (+0.50%)	:arrow_up:
fd/plugin_registry.go	71.87% <66.66%> (+4.22%)	:arrow_up:
pipeline/processor.go	87.94% <40.00%> (-2.01%)	:arrow_down:
plugin/action/debug/debug.go	86.95% <76.92%> (-5.55%)	:arrow_down:
fd/file.d.go	63.55% <26.66%> (+5.56%)	:arrow_up:

... and 34 files with indirect coverage changes

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.