feat: add mqtt docker integration test (#4556)

* feat: add MQTT broker container for integ testing

* fix: spawn-containers was not launching anything for the "all" case

Looks like when the script was refactored to allow for individual
services to be launched, the "all" case was not well tested. As such,
when we didn't name any services to launch nothing would launch.

After fixing this issue, it seems the hdb container doesn't launch
properly in the current state of the script. Something about having
those docker commands in a function or the function being sent to the
background disconnects it from the main tty for the script causing it to
fail to perform the readiness check properly.

In practice it seemed this meant hdb would report "ready" too soon and
the seed data insert would fail, as well as the tests.

For now, I'm skipping hdb. We'll have to loop back to try and find a
good alternative approach.

* test: add test for mqtt.publish

This test currently fails because the client dependency isn't injected.

* chore: inject mqtt dialer for test dependencies

* fix: some containers take time to clean, ignore fails

Sometimes when we run `docker stop` the engine can take a beat before
the container is removed from the active list. The follow-up `docker rm`
can therefore try to destroy the container while it's stopping (and
`rm`'ing itself).

In such a case, ignore the failure. If the container truly sticks around
beyond these 2 commands, we'll see a different failure when we try to
relaunch the container with an identical name.

* chore: print message while NOT launching SAP HANA

* fix: launch SAP HANA without sending job to background

* chore: improve doc for how to shutdown all integ test containers

* chore: make generate

* fix: add span to context for all dependency tests (#4568)

Co-authored-by: Sean Brickley <sean@wabr.io>

Author: Owen Nelson

Makefile

@@ -132,15 +132,15 @@ test-rust:
 	$(CARGO) clippy $(CARGO_ARGS) -- -Dclippy::all -Dclippy::undocumented_unsafe_blocks
 
 INTEGRATION_INJECTION_TESTS=integration_hdb_injection,integration_sqlite_injection,integration_vertica_injection,integration_mssql_injection,integration_mysql_injection,integration_mariadb_injection,integration_pg_injection
-INTEGRATION_WRITE_TESTS=integration_hdb_write_to,integration_sqlite_write_to,integration_vertica_write_to,integration_mssql_write_to,integration_mysql_write_to,integration_mariadb_write_to,integration_pg_write_to
+INTEGRATION_WRITE_TESTS=integration_mqtt_pub,integration_hdb_write_to,integration_sqlite_write_to,integration_vertica_write_to,integration_mssql_write_to,integration_mysql_write_to,integration_mariadb_write_to,integration_pg_write_to
 INTEGRATION_READ_TESTS=integration_hdb_read_from_seed,integration_hdb_read_from_nonseed,integration_sqlite_read_from_seed,integration_sqlite_read_from_nonseed,integration_vertica_read_from_seed,integration_vertica_read_from_nonseed,integration_mssql_read_from_seed,integration_mssql_read_from_nonseed,integration_mariadb_read_from_seed,integration_mariadb_read_from_nonseed,integration_mysql_read_from_seed,integration_mysql_read_from_nonseed,integration_pg_read_from_seed,integration_pg_read_from_nonseed
 INTEGRATION_TESTS="$(INTEGRATION_INJECTION_TESTS),$(INTEGRATION_WRITE_TESTS),$(INTEGRATION_READ_TESTS)"
 
 test-flux:
 	$(GO_RUN) ./cmd/flux test -v --skip $(INTEGRATION_TESTS)
 
 test-flux-integration:
-	./etc/spawn-dbs.sh
+	./etc/spawn-containers.sh
 	$(GO_RUN) ./cmd/flux test -v --test $(INTEGRATION_INJECTION_TESTS)
 	$(GO_RUN) ./cmd/flux test -v --test $(INTEGRATION_WRITE_TESTS)
 	$(GO_RUN) ./cmd/flux test -v --test $(INTEGRATION_READ_TESTS)

dependencies/dependenciestest/dependencies.go

@@ -8,7 +8,9 @@ import (
 	"github.com/influxdata/flux"
 	"github.com/influxdata/flux/dependencies/filesystem"
 	"github.com/influxdata/flux/dependencies/influxdb"
+	"github.com/influxdata/flux/dependencies/mqtt"
 	"github.com/influxdata/flux/dependencies/url"
+	"github.com/influxdata/flux/dependency"
 	"github.com/influxdata/flux/execute"
 	"github.com/influxdata/flux/mock"
 )
@@ -39,10 +41,14 @@ func defaultTestFunction(req *http.Request) *http.Response {
 type Deps struct {
 	flux.Deps
 	influxdb influxdb.Dependency
+	mqtt     mqtt.Dependency
 }
 
 func (d Deps) Inject(ctx context.Context) context.Context {
-	return d.influxdb.Inject(d.Deps.Inject(ctx))
+	ctx = d.Deps.Inject(ctx)
+	ctx = d.influxdb.Inject(ctx)
+	ctx, _ = dependency.Inject(ctx)
+	return d.mqtt.Inject(ctx)
 }
 
 func Default() Deps {
@@ -62,6 +68,9 @@ func Default() Deps {
 		influxdb: influxdb.Dependency{
 			Provider: influxdb.HttpProvider{},
 		},
+		mqtt: mqtt.Dependency{
+			Dialer: mqtt.DefaultDialer{},
+		},
 	}
 }
 

etc/spawn-dbs.sh renamed to etc/spawn-containers.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
-# This script will try to spin up a docker container for a series of database
-# engines which select flux tests will run against.
+# This script will try to spin up a docker container for a series of services
+# (mostly database engines) which select flux tests will run against.
 #
 # Typically you will not invoke this script yourself.
 # Instead use: `make test-flux-integration` from the repo root since the Make
@@ -19,7 +19,12 @@
 # To shutdown all the containers (after you're done running
 # integration tests), you should be able to do something like:
 # ```
-# docker ps --format '{{.Names}}' | grep flux-integ-tests | xargs docker rm -f
+# docker ps --format '{{.Names}}' | grep flux-integ-tests | xargs docker stop
+# ```
+# If you only happen to be using docker _for these integration tests_ you can
+# more succinctly stop everything all at once with:
+# ```
+# docker stop $(docker ps -q)
 # ```
 
 set -e
@@ -34,6 +39,9 @@ MARIADB_NAME="${PREFIX}-mariadb"
 MARIADB_TAG="mariadb:10"
 MS_NAME="${PREFIX}-mssql"
 MS_TAG="mcr.microsoft.com/mssql/server:2019-latest"
+MQTT_TAG="eclipse-mosquitto:2.0.14"
+MQTT_NAME="${PREFIX}-mqtt"
+MQTT_CONFIG_FILE="/tmp/${PREFIX}-mosquitto.conf"
 VERTICA_NAME="${PREFIX}-vertica"
 VERTICA_TAG="vertica/vertica-ce:11.0.0-0"
 SQLITE_DB_PATH="/tmp/${PREFIX}-sqlite.db"
@@ -158,8 +166,11 @@ rm -f "$SQLITE_DB_PATH"
 # volumes that would otherwise be left orphaned.
 # Each container we run _should be_ launched with the `--rm` flag to help cleanup
 # these spent containers as we go.
-docker stop "${HDB_NAME}" "${PG_NAME}" "${MYSQL_NAME}" "${MARIADB_NAME}" "${MS_NAME}" "${VERTICA_NAME}" \
-|| docker rm -f "${HDB_NAME}" "${PG_NAME}" "${MYSQL_NAME}" "${MARIADB_NAME}" "${MS_NAME}" "${VERTICA_NAME}"
+docker stop \
+  "${HDB_NAME}" "${PG_NAME}" "${MYSQL_NAME}" "${MARIADB_NAME}" "${MS_NAME}" "${VERTICA_NAME}" "${MQTT_NAME}" \
+|| docker rm -f \
+  "${HDB_NAME}" "${PG_NAME}" "${MYSQL_NAME}" "${MARIADB_NAME}" "${MS_NAME}" "${VERTICA_NAME}" "${MQTT_NAME}" \
+|| true
 
 function wait_for () {
   name="${1}"
@@ -259,10 +270,31 @@ function run_pg {
   docker exec "${PG_NAME}" psql -U postgres -c "${PG_SEED}"
 }
 
+function run_mqtt {
+  cat <<'EOF' > "${MQTT_CONFIG_FILE}"
+listener 1883
+allow_anonymous true
+EOF
+
+  docker run --rm --detach \
+    --name "${MQTT_NAME}" \
+    --publish 1883:1883 \
+    -v "${MQTT_CONFIG_FILE}:/mosquitto/config/mosquitto.conf:ro" \
+    "${MQTT_TAG}" \
+     mosquitto -v -c /mosquitto/config/mosquitto.conf
+
+  wait_for "MQTT" "docker exec ${MQTT_NAME} mosquitto_pub -t 'liveness' -n"
+  # XXX: the flux mqtt support currently only exposes write operations.
+  # If we also supported reads, we could seed the queue here with some retained
+  # messages or by setting the QoS value.
+}
+
+# trailing args to the script can be service names, checked by `should_start`
+TAIL="$*"
 function should_start {
-    # Start a databases if no the script is invoked without any arguments, or if the database is
-    # among the arguments
-    [ "$1" == "" ] || (echo "${@:2}" | grep "$1")
+    # Start a container if no the script is invoked without any arguments, or if
+    # the service is among the arguments
+    [ "${TAIL}" == "" ] || (echo "$TAIL" | grep "$1")
 }
 
 if should_start "mariadb" "$@" ; then
@@ -285,11 +317,6 @@ if should_start "vertica" "$@" ; then
     run_vertica &
 fi
 
-if should_start "saphana" "$@" ; then
-    echo "Starting SAP Hana"
-    run_sap_hana &
-fi
-
 if should_start "postgres" "$@" ; then
     echo "Starting Postgres"
     run_pg &
@@ -300,4 +327,16 @@ if should_start "sqlite" "$@" ; then
     sqlite3 "${SQLITE_DB_PATH}" "${SQLITE_SEED}" &
 fi
 
+if should_start "mqtt" "$@" ; then
+    echo "Starting MQTT"
+    run_mqtt &
+fi
+
+if should_start "saphana" "$@" ; then
+    echo "Starting SAP Hana"
+    # n.b. this relies on docker commands that bind tty (via the `-it` flags).
+    # This means it cannot be backgrounded like the others (with `&`).
+    run_sap_hana
+fi
+
 wait

libflux/go/libflux/buildinfo.gen.go

@@ -201,6 +201,7 @@ var sourceHashes = map[string]string{
 	"stdlib/experimental/min_test.flux":                                                           "9c50f8ed865bc4ce29596c58be44e76071be9ca16ef57d65156d563e920ea922",
 	"stdlib/experimental/mode_test.flux":                                                          "e125d0d2de23a705e3593ce8f414a3f3519b852083ebd989748ab7dd5d203257",
 	"stdlib/experimental/mqtt/mqtt.flux":                                                          "31aeae56a961429a0a69f930f2184a006b6f4269923d73dde98883fa5bec8805",
+	"stdlib/experimental/mqtt/mqtt_test.flux":                                                     "bfb77c41bccffcf5890027ec49e29c9c607c7bb59459dabec772fb989031c5a9",
 	"stdlib/experimental/oee/apq_test.flux":                                                       "6a64721efd123bbfd40ca90cbebc7cfd2dccbd651da5e1ee4d33e0833256be60",
 	"stdlib/experimental/oee/computeapq_test.flux":                                                "a5069718fb11cee0ffc8b345ead279ccb7f4bcc4644ffdc8d641c310be1969ec",
 	"stdlib/experimental/oee/oee.flux":                                                            "30d6ce4144d1425cb9b294bd17fd73395d2a7c87b86c089aa89e73969a5b98f3",

stdlib/experimental/mqtt/flux_test_gen.go

@@ -0,0 +1,16 @@
+// DO NOT EDIT: This file is autogenerated via the builtin command.
+
+package mqtt
+
+import ast "github.com/influxdata/flux/ast"
+
+var FluxTestPackages = []*ast.Package{&ast.Package{
+	BaseNode: ast.BaseNode{
+		Comments: nil,
+		Errors:   nil,
+		Loc:      nil,
+	},
+	Files:   []*ast.File{},
+	Package: "mqtt_test",
+	Path:    "experimental/mqtt",
+}}

stdlib/experimental/mqtt/mqtt_test.flux

@@ -0,0 +1,28 @@
+package mqtt_test
+
+
+import "array"
+import "experimental/mqtt"
+import "testing"
+
+testcase integration_mqtt_pub {
+    got =
+        array.from(
+            rows: [
+                {
+                    ok:
+                        mqtt.publish(
+                            broker: "tcp://127.0.0.1:1883",
+                            topic: "test/topic",
+                            message: "smoke test",
+                            qos: 0,
+                            retain: false,
+                            clientid: "fluxtest",
+                        ),
+                },
+            ],
+        )
+    want = array.from(rows: [{ok: true}])
+
+    testing.diff(want: want, got: got) |> yield()
+}

stdlib/test_packages.go

@@ -21,6 +21,7 @@ import (
 	bitwise "github.com/influxdata/flux/stdlib/experimental/bitwise"
 	geo "github.com/influxdata/flux/stdlib/experimental/geo"
 	json "github.com/influxdata/flux/stdlib/experimental/json"
+	mqtt "github.com/influxdata/flux/stdlib/experimental/mqtt"
 	oee "github.com/influxdata/flux/stdlib/experimental/oee"
 	prometheus "github.com/influxdata/flux/stdlib/experimental/prometheus"
 	record "github.com/influxdata/flux/stdlib/experimental/record"
@@ -73,6 +74,7 @@ var FluxTestPackages = func() []*ast.Package {
 	pkgs = append(pkgs, bitwise.FluxTestPackages...)
 	pkgs = append(pkgs, geo.FluxTestPackages...)
 	pkgs = append(pkgs, json.FluxTestPackages...)
+	pkgs = append(pkgs, mqtt.FluxTestPackages...)
 	pkgs = append(pkgs, oee.FluxTestPackages...)
 	pkgs = append(pkgs, prometheus.FluxTestPackages...)
 	pkgs = append(pkgs, record.FluxTestPackages...)