Commit Graph

204 Commits

Author SHA1 Message Date
Thales Macedo Garitezi a78ecc4bb6
Merge pull request #9579 from thalesmg/bugfix-kafka-producer-stop-client-ee50
fix(kafka_producer): cleanup client after failing to start producers
2022-12-26 08:59:36 -03:00
Zaiming (Stone) Shi 59970272ff chore: pin hocon 0.33.0 2022-12-23 12:07:36 +01:00
Erik Timan 13942f5c49 refactor: rename error return in mysql connector 2022-12-22 10:29:12 +01:00
Erik Timan aab914d65a test: refactor EE mysql bridge test case
Co-authored-by: Thales Macedo Garitezi <thalesmg@gmail.com>
2022-12-22 10:27:38 +01:00
Erik Timan 7df24000a0 test: add more EE mysql bridge test cases 2022-12-22 10:27:38 +01:00
Thales Macedo Garitezi 7242ce426b fix(kafka_producer): cleanup client after failing to start producers
https://emqx.atlassian.net/browse/EMQX-8547

If a Kafka Producer bridge is given bad configuration (e.g.: bad authn
credentials), the Wolff client process is started successfully, as it
does not attempt to connect, but when the producer process is
attempted to be started, it fails (only then the client tries to
connect to Kafka).  At this point, an error was thrown, but the
supervised client process remained running.

If the configuration was later fixed and the bridge updated, which
prompted its removal and recreation, the Wolff client would report to
be "already started", so it would never pick up the new (fixed)
configuration, and the producers would perpetually fail to start until
the node would be restarted.

We simply ensure the client is stopped before throwing the error,
unrolling the start-up procedure.
2022-12-19 17:48:40 -03:00
Thales Macedo Garitezi c0b208dd9e fix(influxdb): check if fields are empty before sending
Related Issue: https://emqx.atlassian.net/browse/EMQX-8461

Currently, the InfluxDB client raises an error if an empty `fields`
map is passed to it for pushing data.

```
  14:03:35.563 [error] [InfluxDB] Encode [
    %{
      fields: %{},
      measurement: "t/topic",
      tags: %{},
      timestamp: 1670864615563
    }
  ] failed: :error :missing_field [
    {:influxdb_line, :encode_fields, 1,
     [
       file: '/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb_line.erl',
       line: 60
     ]},
    {:influxdb_line, :encode_, 1,
     [
       file: '/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb_line.erl',
       line: 44
     ]},
    {:influxdb_line, :"-encode_/1-fun-0-", 2,
     [
       file: '/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb_line.erl',
       line: 38
     ]},
    {:influxdb, :write, 2,
     [file: '/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb.erl', line: 79]},
    {:emqx_ee_connector_influxdb, :do_query, 3,
     [
       file: '/home/thales/dev/emqx/emqx/lib-ee/emqx_ee_connector/src/emqx_ee_connector_influxdb.erl',
       line: 322
     ]},
    {:emqx_resource_worker, :apply_query_fun, 6,
     [
       file: '/home/thales/dev/emqx/emqx/apps/emqx_resource/src/emqx_resource_worker.erl',
       line: 454
     ]},
    {:emqx_resource_worker, :query_or_acc, 3,
     [
       file: '/home/thales/dev/emqx/emqx/apps/emqx_resource/src/emqx_resource_worker.erl',
       line: 306
     ]},
    {:gen_statem, :loop_state_callback, 11, [file: 'gen_statem.erl', line: 1203]},
    {:proc_lib, :init_p_do_apply, 3, [file: 'proc_lib.erl', line: 226]}
  ]
  2022-12-12T14:03:35.563607-03:00 [error] [InfluxDB] Encode [#{fields => #{},measurement => <<"t/topic">>,tags => #{},timestamp => 1670864615563}] failed: error missing_field [{influxdb_line,encode_fields,1,[{file,"/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb_line.erl"},{line,60}]},{influxdb_line,encode_,1,[{file,"/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb_line.erl"},{line,44}]},{influxdb_line,'-encode_/1-fun-0-',2,[{file,"/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb_line.erl"},{line,38}]},{influxdb,write,2,[{file,"/home/thales/dev/emqx/emqx/deps/influxdb/src/influxdb.erl"},{line,79}]},{emqx_ee_connector_influxdb,do_query,3,[{file,"/home/thales/dev/emqx/emqx/lib-ee/emqx_ee_connector/src/emqx_ee_connector_influxdb.erl"},{line,322}]},{emqx_resource_worker,apply_query_fun,6,[{file,"/home/thales/dev/emqx/emqx/apps/emqx_resource/src/emqx_resource_worker.erl"},{line,454}]},{emqx_resource_worker,query_or_acc,3,[{file,"/home/thales/dev/emqx/emqx/apps/emqx_resource/src/emqx_resource_worker.erl"},{line,306}]},{gen_statem,loop_state_callback,11,[{file,"gen_statem.erl"},{line,1203}]},{proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,226}]}]

  14:03:35.565 [error] [connector: "bridge:influxdb_api_v2:aaa", msg: 'influxdb write point failed', reason: :missing_field]
  2022-12-12T14:03:35.565345-03:00 [error] connector: <<"bridge:influxdb_api_v2:aaa">>, line: 335, mfa: emqx_ee_connector_influxdb:do_query/3, msg: influxdb write point failed, reason: missing_field
  []

  14:03:35.565 [error] [id: "bridge:influxdb_api_v2:aaa", msg: :send_error, reason: :missing_field]
  iex(emqx@127.0.0.1)2> 2022-12-12T14:03:35.565913-03:00 [error] id: <<"bridge:influxdb_api_v2:aaa">>, line: 396, mfa: emqx_resource_worker:handle_query_result/4, msg: send_error, reason: missing_field
```

Instead of raising, we check if the interpolation resulted in an empty
map due to lack of context and just fail the push more gracefully.

Related to this, the original issue _appears_ to be related to a
frontend issue (to be confirmed and fixed separately), where the
it is not encoding the field types:
https://emqx.atlassian.net/browse/EMQX-8461?focusedCommentId=24805
2022-12-16 09:52:34 -03:00
Zaiming (Stone) Shi 9e3da5b661 chore: bump app versions 2022-12-14 20:07:41 +01:00
Zaiming (Stone) Shi 56066a03b5 Merge remote-tracking branch 'origin/release-50' into 1214-sync-master-upstreams 2022-12-14 20:04:20 +01:00
Zaiming (Stone) Shi 34f7fff97d chore: pin hocon 0.32.0 2022-12-14 15:53:42 +01:00
Zaiming (Stone) Shi 42c58e2a91 Merge remote-tracking branch 'origin/release-50' into 1214-sync-master-upstreams 2022-12-14 15:29:13 +01:00
Thales Macedo Garitezi 2932aceaf9 refactor: rename and move `emqx_ee_bridge_gcp_pubsub_resource` to `emqx_ee_connector_gcp_pubsub` 2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi 464d0a5057 refactor(test): use a linked janitor for test teardown 2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi b66d2904be docs: improve descriptions
Co-authored-by: Zaiming (Stone) Shi <zmstone@gmail.com>
2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi b9bc82f87a feat(gcp_pubsub): add `local_topic` config
Given the implicit convention that an egress bridge containing the
`local_topic` config will forward messages without the need for a rule
action, this was added to avoid needing a rule action.
2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi ac048dbafa docs(gcp_pubsub): describe where to get credentials 2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi a095867358 test(refactor): add ids to namespace `on_exit` callbacks 2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi 697b3ecf2d docs: improve descriptions
Co-authored-by: Zaiming (Stone) Shi <zmstone@gmail.com>
2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi 1cd91a24e9 feat(gcp_pubsub): implement GCP PubSub bridge (ee5.0) 2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi 82be9d878d test(flaky): avoid inter-suite flakiness 2022-12-12 17:18:19 -03:00
Thales Macedo Garitezi c69022f3c4 test(ci): clean config after test
Trying to fix this error for profile `emqx`:

```
=CRASH REPORT==== 30-Nov-2022::13:25:46.763989 ===
  crasher:
    initial call: application_master:init/4
    pid: <0.9682.1>
    registered_name: []
    exception exit: {bad_return,
                        {{emqx_conf_app,start,[normal,[]]},
                         {emqx_conf_schema,
                             [#{kind => validation_error,path => "bridges",
                                reason => unknown_fields,
                                unknown => <<"influxdb_api_v1">>,
                                unmatched => <<"mqtt,webhook">>}]}}}
      in function  application_master:init/4 (application_master.erl, line 142)
    ancestors: [<0.9681.1>]
    message_queue_len: 1
    messages: [{'EXIT',<0.9683.1>,normal}]
    links: [<0.9681.1>,<0.44.0>]
    dictionary: []
    trap_exit: true
    status: running
    heap_size: 610
    stack_size: 29
    reductions: 195
  neighbours:

=INFO REPORT==== 30-Nov-2022::13:25:46.777895 ===
    application: emqx_conf
    exited: {bad_return,
                {{emqx_conf_app,start,[normal,[]]},
                 {emqx_conf_schema,
                     [#{kind => validation_error,path => "bridges",
                        reason => unknown_fields,
                        unknown => <<"influxdb_api_v1">>,
                        unmatched => <<"mqtt,webhook">>}]}}}
    type: temporary

%%% emqx_plugins_SUITE ==> init_per_suite: FAILED
%%% emqx_plugins_SUITE ==> {{failed_to_start_app,emqx_conf,
     {emqx_conf,
         {bad_return,
             {{emqx_conf_app,start,[normal,[]]},
              {emqx_conf_schema,
                  [#{kind => validation_error,path => "bridges",
                     reason => unknown_fields,
                     unknown => <<"influxdb_api_v1">>,
                     unmatched => <<"mqtt,webhook">>}]}}}}},
 [{emqx_common_test_helpers,start_app,4,
      [{file,
           "/__w/emqx/emqx/source/apps/emqx/test/emqx_common_test_helpers.erl"},
       {line,227}]},
  {lists,foreach,2,[{file,"lists.erl"},{line,1342}]},
  {emqx_plugins_SUITE,init_per_suite,1,
      [{file,
           "/__w/emqx/emqx/source/apps/emqx_plugins/test/emqx_plugins_SUITE.erl"},
       {line,34}]},
  {test_server,ts_tc,3,[{file,"test_server.erl"},{line,1783}]},
  {test_server,run_test_case_eval1,6,[{file,"test_server.erl"},{line,1380}]},
  {test_server,run_test_case_eval,9,[{file,"test_server.erl"},{line,1224}]}]}

%%% emqx_plugins_SUITE ==> t_bad_info_json: SKIPPED
%%% emqx_plugins_SUITE ==> {tc_auto_skip,
    {failed,
        {emqx_plugins_SUITE,init_per_suite,
            {'EXIT',
                {{failed_to_start_app,emqx_conf,
                     {emqx_conf,
                         {bad_return,
                             {{emqx_conf_app,start,[normal,[]]},
                              {emqx_conf_schema,
                                  [#{kind => validation_error,
                                     path => "bridges",
                                     reason => unknown_fields,
                                     unknown => <<"influxdb_api_v1">>,
                                     unmatched => <<"mqtt,webhook">>}]}}}}},
                 [{emqx_common_test_helpers,start_app,4,
                      [{file,
                           "/__w/emqx/emqx/source/apps/emqx/test/emqx_common_test_helpers.erl"},
                       {line,227}]},
                  {lists,foreach,2,[{file,"lists.erl"},{line,1342}]},
                  {emqx_plugins_SUITE,init_per_suite,1,
                      [{file,
                           "/__w/emqx/emqx/source/apps/emqx_plugins/test/emqx_plugins_SUITE.erl"},
                       {line,34}]},
                  {test_server,ts_tc,3,[{file,"test_server.erl"},{line,1783}]},
                  {test_server,run_test_case_eval1,6,
                      [{file,"test_server.erl"},{line,1380}]},
                  {test_server,run_test_case_eval,9,
                      [{file,"test_server.erl"},{line,1224}]}]}}}}}
```
2022-12-12 17:18:13 -03:00
Thales Macedo Garitezi 34e9056779 refactor: fix typo in variable name
Might confuse people to think it's related to `replayq`.
2022-12-12 17:17:51 -03:00
Zaiming (Stone) Shi 40809b2ad0 Merge remote-tracking branch 'origin/dev/ee5.0' into release-50 2022-12-09 11:45:52 +01:00
Zaiming (Stone) Shi dc14cd450d test: render config for emqx_conf only for ee bridge tests 2022-12-07 18:27:06 +01:00
Ilya Averyanov 6692b0c895 feat(bridge): add Redis bridge 2022-12-06 23:15:42 +03:00
Erik Timan 485c4ddcab
Merge pull request #9393 from olcai/add-mysql-bridge-test
test: add basic mysql bridge test (ee5.0)
2022-12-02 09:43:22 +01:00
Zaiming (Stone) Shi 9ea22d062d refactor: make all bridges optional (required = false) 2022-12-01 16:50:10 +01:00
Zaiming (Stone) Shi 7adb539ae1 chore: pin hocon 0.31.2 2022-12-01 16:50:10 +01:00
Erik Timan 499a32ce36 test: remove unnecessary async tests in EE mysql bridge
The async test cases is not needed since the mysql connector uses the
always_sync callback mode.
2022-12-01 16:38:03 +01:00
Erik Timan 5461311edc test: more EE mysql bridge tests 2022-12-01 10:27:51 +01:00
Erik Timan eb62192838 test: expand EE mysql bridge test with toxiproxy 2022-12-01 10:27:51 +01:00
Erik Timan 96bff1d32e test: improve basic tests for EE mysql bridge 2022-12-01 10:27:51 +01:00
Erik Timan 3e679ceb57 test: add basic tests for EE mysql bridge 2022-12-01 10:27:51 +01:00
Zaiming (Stone) Shi fa90638de2 chore: add description to apps 2022-11-30 17:03:41 +01:00
Zaiming (Stone) Shi 9382399728 chore: add an empty file emqx_license.conf to make ci happy 2022-11-30 12:52:26 +01:00
Zaiming (Stone) Shi 83fe35b148 Merge remote-tracking branch 'origin/dev/ee5.0' into 1116-sync-v5.0.10-to-ee50 2022-11-18 19:23:31 +01:00
Zaiming (Stone) Shi 7305923d12 fix: bridge name parser should not leak atom 2022-11-16 18:25:42 +01:00
Zaiming (Stone) Shi c940b901f5 chore: fix app versions 2022-11-16 16:26:43 +01:00
Thales Macedo Garitezi 0f9cc0d93f test(refactor): stop snabbkaffe on every test 2022-11-11 10:03:30 -03:00
Thales Macedo Garitezi 04588148b7 test(influxdb): increase influxdb bridge/connector coverage (ee5.0) 2022-11-07 15:15:49 -03:00
Thales Macedo Garitezi ee4c723fcb refactor: simplify wolff telemetry handler id 2022-10-17 16:26:04 -03:00
Thales Macedo Garitezi 1b2b629cdd feat: emit telemetry events for all resource worker metrics 2022-10-13 15:32:04 -03:00
Thales Macedo Garitezi 1ad3b5df17 fix: uninstall telemetry handler on resource stop, use unique id 2022-10-13 10:29:53 -03:00
Thales Macedo Garitezi 4475289ce4 feat: use upstream newly tagged 1.7.0 wolff 2022-10-11 09:47:40 -03:00
Thales Macedo Garitezi 98500313eb fix(kafka): some fixes for kafka producer
- MQTT topic should be a binary
- use correct gauge functions from `wolff_metrics`.
- don't double increment success counter for kafka action
- adds a few more metrics assertions
2022-10-10 17:11:29 -03:00
Kjell Winblad 57270fb8fc feat: add support for counters and gauges to the Kafka Bridge
This commit adds support for counters and gauges to the Kafka Brige.
The Kafka bridge uses [Wolff](https://github.com/kafka4beam/wolff) for
the  Kafka connection. Wolff does its own batching and does not use the
batching functionality in `emqx_resource_worker` that is used by other
bridge types. Therefore, the counter events have to be generated by
Wolff. We have added
[telemetry](https://github.com/beam-telemetry/telemetry) events to Wolff
that we hook into to change counters and gauges for the Kafka bridge. The
counter called `matched` does not depend on specific functionality of
any bridge type so the updates of this counter is moved higher up in the
call chain then previously so that it also gets updated for Kafka
bridges.
2022-10-10 14:40:57 -03:00
Shawn dfe14be8b2 chore: bump app vsns 2022-09-26 18:23:28 +08:00
Shawn 4135910b42 chore: merge master into dev/ee5.0 2022-09-26 09:52:33 +08:00
Zaiming (Stone) Shi f6ac4c3a76
Merge pull request #8798 from zmstone/0815-feat-add-kafka-connector
feat: Add Kafka connector
2022-09-24 22:57:50 +02:00
Kjell Winblad a3c88b40a0 test: changes to make Kafka container run in GitHub action 2022-09-23 14:33:41 +02:00