mirror of
https://github.com/absmach/magistrala.git
synced 2026-08-07 07:14:46 +00:00
NOISSUE - Use internal principals for services communication (#3564)
Signed-off-by: dusan <borovcanindusan1@gmail.com>
This commit is contained in:
Dušan Borovčanin
GitHub
@@ -197,7 +197,14 @@ FILTERED_SERVICES = $(filter-out $(RUN_ADDON_ARGS), $(SERVICES))
|
||||
|
||||
all: $(SERVICES)
|
||||
|
||||
.PHONY: all $(SERVICES) dockers dockers_dev latest release provision_atom_tokens provision-atom-tokens migrate_atom run_latest run_latest_ci run_tls run_stable run_addons grpc_mtls_certs check_mtls check_certs check_fluxmq_service_certs check_re_trace_key test_api mocks
|
||||
.PHONY: all help $(SERVICES) dockers dockers_dev latest release provision_atom_tokens provision-atom-tokens migrate_atom run_latest run_latest_ci run_tls run_stable run_addons grpc_mtls_certs check_mtls check_certs check_fluxmq_service_certs check_re_trace_key test_api mocks
|
||||
|
||||
help:
|
||||
@printf 'Usage:\n make <target> [VARIABLE=value ...]\n\nAvailable targets:\n'
|
||||
@$(MAKE) -qpRr : 2>/dev/null | \
|
||||
awk -F: '/^[[:alnum:]_][^$$#\/\t=]*:([^=]|$$)/ { split($$1, targets, /[[:space:]]+/); for (i in targets) if (targets[i] != "") print targets[i] }' | \
|
||||
LC_ALL=C sort -u | \
|
||||
awk '$$0 != "Makefile" { printf " make %s\n", $$0 }'
|
||||
|
||||
clean:
|
||||
rm -rf ${BUILD_DIR}
|
||||
@@ -349,16 +356,28 @@ else
|
||||
$(eval GRPC_MTLS :=)
|
||||
endif
|
||||
|
||||
# The Rules Engine reaches FluxMQ on the mTLS local listener, and compose mounts
|
||||
# both sides of that connection. The certificates and the principal secret are
|
||||
# Internal services reach FluxMQ on the mTLS local listener, and Compose mounts
|
||||
# both sides of those connections. Certificates and principal secrets are
|
||||
# generated rather than committed, so make them before anything binds them.
|
||||
check_fluxmq_service_certs:
|
||||
ifeq ("$(wildcard docker/ssl/certs/re-fluxmq-client.crt)","")
|
||||
$(MAKE) -C docker/ssl fluxmq_service_certs
|
||||
endif
|
||||
ifeq ("$(wildcard docker/ssl/certs/timescale-writer-fluxmq-client.crt)","")
|
||||
$(MAKE) -C docker/ssl timescale_writer_fluxmq_client_cert
|
||||
endif
|
||||
ifeq ("$(wildcard docker/ssl/certs/postgres-writer-fluxmq-client.crt)","")
|
||||
$(MAKE) -C docker/ssl postgres_writer_fluxmq_client_cert
|
||||
endif
|
||||
ifeq ("$(wildcard docker/fluxmq/secrets/re-current)","")
|
||||
$(MAKE) -C docker/ssl fluxmq_service_secret
|
||||
endif
|
||||
ifeq ("$(wildcard docker/fluxmq/secrets/timescale-writer-current)","")
|
||||
$(MAKE) -C docker/ssl timescale_writer_fluxmq_service_secret
|
||||
endif
|
||||
ifeq ("$(wildcard docker/fluxmq/secrets/postgres-writer-current)","")
|
||||
$(MAKE) -C docker/ssl postgres_writer_fluxmq_service_secret
|
||||
endif
|
||||
|
||||
check_re_trace_key:
|
||||
ifeq ("$(wildcard docker/re/secrets/trace.key)","")
|
||||
|
||||
@@ -300,34 +300,54 @@ make check_certs
|
||||
|
||||
This creates whatever is missing and leaves anything already present alone:
|
||||
|
||||
| Path | What it is |
|
||||
| -------------------------------------------------- | ----------------------------------------------------------------------------- |
|
||||
| `docker/ssl/certs/fluxmq-service-server.{crt,key}` | Server certificate for FluxMQ's mTLS service listener |
|
||||
| `docker/ssl/certs/re-fluxmq-client.{crt,key}` | Client certificate whose URI SAN identifies the Rules Engine to that listener |
|
||||
| `docker/fluxmq/secrets/re-current` | The principal secret, derived from `MG_RE_BROKER_SECRET` in `docker/.env` |
|
||||
| `docker/re/secrets/trace.key` | HMAC key the Rules Engine signs its loop-detection traces with |
|
||||
| Path | What it is |
|
||||
| ----------------------------------------------------------- | ------------------------------------------------------------------------- |
|
||||
| `docker/ssl/certs/fluxmq-service-server.{crt,key}` | Server certificate for FluxMQ's mTLS service listener |
|
||||
| `docker/ssl/certs/re-fluxmq-client.{crt,key}` | Client certificate whose URI SAN identifies the Rules Engine |
|
||||
| `docker/ssl/certs/timescale-writer-fluxmq-client.{crt,key}` | Client certificate whose URI SAN identifies the Timescale writer |
|
||||
| `docker/ssl/certs/postgres-writer-fluxmq-client.{crt,key}` | Client certificate whose URI SAN identifies the Postgres writer |
|
||||
| `docker/fluxmq/secrets/re-current` | Rules Engine principal secret, from `MG_RE_BROKER_SECRET` |
|
||||
| `docker/fluxmq/secrets/timescale-writer-current` | Timescale writer secret, from `MG_TIMESCALE_WRITER_BROKER_SECRET` |
|
||||
| `docker/fluxmq/secrets/postgres-writer-current` | Postgres writer secret, from `MG_POSTGRES_WRITER_BROKER_SECRET` |
|
||||
| `docker/re/secrets/trace.key` | HMAC key the Rules Engine signs its loop-detection traces with |
|
||||
|
||||
Both certificates are issued by the development CA committed at
|
||||
Internal services reach the broker as *local principals* rather than as ordinary
|
||||
clients: each presents a client certificate whose URI SAN names it, plus a SASL
|
||||
secret, and the broker grants it only what it needs — the Rules Engine consumes
|
||||
`m` and republishes under it, the writers only subscribe to `writers`. The
|
||||
principals are declared in `docker/fluxmq/node{1,2,3}.yaml`, and adding a service
|
||||
means adding an entry there alongside its certificate and secret.
|
||||
|
||||
The certificates are issued by the development CA committed at
|
||||
`docker/ssl/certs/ca.crt`, so no extra setup is needed for a local run. The
|
||||
generated material is gitignored.
|
||||
|
||||
The server certificate is issued for `fluxmq` and `fluxmq-node{1,2,3}`, which
|
||||
covers both this Compose stack and a single-node deployment. Point
|
||||
`MG_RE_BROKER_URL` at a host outside that set and the Rules Engine fails its
|
||||
TLS verification with `certificate is valid for ...`; add the name to
|
||||
covers both this Compose stack and a single-node deployment. Point any
|
||||
`MG_*_BROKER_URL` at a host outside that set and the service fails its TLS
|
||||
verification with `certificate is valid for ...`; add the name to
|
||||
`FLUXMQ_SERVICE_SERVER_CERT_CONFIG` in `docker/ssl/Makefile` and reissue:
|
||||
|
||||
```bash
|
||||
rm -f docker/ssl/certs/fluxmq-service-server.* docker/ssl/certs/re-fluxmq-client.*
|
||||
rm -f docker/ssl/certs/fluxmq-service-server.* \
|
||||
docker/ssl/certs/re-fluxmq-client.* \
|
||||
docker/ssl/certs/timescale-writer-fluxmq-client.* \
|
||||
docker/ssl/certs/postgres-writer-fluxmq-client.*
|
||||
make -C docker/ssl fluxmq_service_certs
|
||||
```
|
||||
|
||||
`make check_certs` skips this pair when it already exists, so a stale
|
||||
certificate has to be removed rather than merely re-run.
|
||||
`make check_certs` skips certificates that already exist, so stale certificates
|
||||
have to be removed rather than merely re-running the target.
|
||||
|
||||
Two of these are stateful, not merely derived. `re-current` must stay equal to
|
||||
`MG_RE_BROKER_SECRET`: change the variable and re-run `make -C docker/ssl
|
||||
fluxmq_service_secret`, or the Rules Engine fails to authenticate.
|
||||
Each local-principal secret must stay equal to the corresponding value in
|
||||
`docker/.env`; a mismatch fails that service's broker authentication. After
|
||||
changing one, re-run its target:
|
||||
|
||||
| Variable | Target |
|
||||
| ----------------------------------- | ----------------------------------------- |
|
||||
| `MG_RE_BROKER_SECRET` | `fluxmq_service_secret` |
|
||||
| `MG_TIMESCALE_WRITER_BROKER_SECRET` | `timescale_writer_fluxmq_service_secret` |
|
||||
| `MG_POSTGRES_WRITER_BROKER_SECRET` | `postgres_writer_fluxmq_service_secret` |
|
||||
`trace.key` is created once and preserved on later runs — replacing it while
|
||||
messages are in flight would invalidate the rule traces they already carry, so
|
||||
delete it only deliberately. Every Rules Engine replica must read the same key.
|
||||
|
||||
+23
-10
@@ -20,8 +20,9 @@ import (
|
||||
writerpg "github.com/absmach/magistrala/consumers/writers/postgres"
|
||||
mglog "github.com/absmach/magistrala/logger"
|
||||
jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
|
||||
"github.com/absmach/magistrala/pkg/messaging"
|
||||
brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
|
||||
brokers "github.com/absmach/magistrala/pkg/messaging/writers"
|
||||
"github.com/absmach/magistrala/pkg/messaging/writers"
|
||||
pgclient "github.com/absmach/magistrala/pkg/postgres"
|
||||
"github.com/absmach/magistrala/pkg/prometheus"
|
||||
"github.com/absmach/magistrala/pkg/server"
|
||||
@@ -41,13 +42,16 @@ const (
|
||||
)
|
||||
|
||||
type config struct {
|
||||
LogLevel string `env:"MG_POSTGRES_WRITER_LOG_LEVEL" envDefault:"info"`
|
||||
ConfigPath string `env:"MG_POSTGRES_WRITER_CONFIG_PATH" envDefault:"/config.toml"`
|
||||
BrokerURL string `env:"MG_MESSAGE_BROKER_URL" envDefault:"nats://localhost:4222"`
|
||||
JaegerURL url.URL `env:"MG_JAEGER_URL" envDefault:"http://localhost:4318/v1/traces"`
|
||||
SendTelemetry bool `env:"MG_SEND_TELEMETRY" envDefault:"true"`
|
||||
InstanceID string `env:"MG_POSTGRES_WRITER_INSTANCE_ID" envDefault:""`
|
||||
TraceRatio float64 `env:"MG_JAEGER_TRACE_RATIO" envDefault:"1.0"`
|
||||
LogLevel string `env:"MG_POSTGRES_WRITER_LOG_LEVEL" envDefault:"info"`
|
||||
ConfigPath string `env:"MG_POSTGRES_WRITER_CONFIG_PATH" envDefault:"/config.toml"`
|
||||
BrokerURL string `env:"MG_MESSAGE_BROKER_URL" envDefault:"nats://localhost:4222"`
|
||||
BrokerClientCert string `env:"MG_POSTGRES_WRITER_BROKER_CLIENT_CERT" envDefault:""`
|
||||
BrokerClientKey string `env:"MG_POSTGRES_WRITER_BROKER_CLIENT_KEY" envDefault:""`
|
||||
BrokerCACerts string `env:"MG_POSTGRES_WRITER_BROKER_CA_CERTS" envDefault:""`
|
||||
JaegerURL url.URL `env:"MG_JAEGER_URL" envDefault:"http://localhost:4318/v1/traces"`
|
||||
SendTelemetry bool `env:"MG_SEND_TELEMETRY" envDefault:"true"`
|
||||
InstanceID string `env:"MG_POSTGRES_WRITER_INSTANCE_ID" envDefault:""`
|
||||
TraceRatio float64 `env:"MG_JAEGER_TRACE_RATIO" envDefault:"1.0"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -107,7 +111,16 @@ func main() {
|
||||
}()
|
||||
tracer := tp.Tracer(svcName)
|
||||
|
||||
pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
|
||||
// The mTLS client identity is optional: brokers that expose an unauthenticated
|
||||
// listener need none. Passing the option with empty paths would fail the
|
||||
// connection outright, so only ask for it once one of them is configured. A
|
||||
// partial configuration still fails, since half an identity is not a usable one.
|
||||
var brokerOpts []messaging.Option
|
||||
if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
|
||||
brokerOpts = append(brokerOpts, writers.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
|
||||
}
|
||||
|
||||
pubSub, err := writers.NewPubSub(ctx, cfg.BrokerURL, logger, brokerOpts...)
|
||||
if err != nil {
|
||||
logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
|
||||
exitCode = 1
|
||||
@@ -119,7 +132,7 @@ func main() {
|
||||
repo := newService(db, logger)
|
||||
repo = consumertracing.NewBlocking(tracer, repo, httpServerConfig)
|
||||
|
||||
if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, brokers.AllTopic, logger); err != nil {
|
||||
if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, writers.AllTopic, logger); err != nil {
|
||||
logger.Error(fmt.Sprintf("failed to create Postgres writer: %s", err))
|
||||
exitCode = 1
|
||||
return
|
||||
|
||||
@@ -20,8 +20,9 @@ import (
|
||||
"github.com/absmach/magistrala/consumers/writers/timescale"
|
||||
mglog "github.com/absmach/magistrala/logger"
|
||||
jaegerclient "github.com/absmach/magistrala/pkg/jaeger"
|
||||
"github.com/absmach/magistrala/pkg/messaging"
|
||||
brokerstracing "github.com/absmach/magistrala/pkg/messaging/brokers/tracing"
|
||||
brokers "github.com/absmach/magistrala/pkg/messaging/writers"
|
||||
"github.com/absmach/magistrala/pkg/messaging/writers"
|
||||
pgclient "github.com/absmach/magistrala/pkg/postgres"
|
||||
"github.com/absmach/magistrala/pkg/prometheus"
|
||||
"github.com/absmach/magistrala/pkg/server"
|
||||
@@ -41,13 +42,16 @@ const (
|
||||
)
|
||||
|
||||
type config struct {
|
||||
LogLevel string `env:"MG_TIMESCALE_WRITER_LOG_LEVEL" envDefault:"info"`
|
||||
ConfigPath string `env:"MG_TIMESCALE_WRITER_CONFIG_PATH" envDefault:"/config.toml"`
|
||||
BrokerURL string `env:"MG_MESSAGE_BROKER_URL" envDefault:"nats://localhost:4222"`
|
||||
JaegerURL url.URL `env:"MG_JAEGER_URL" envDefault:"http://localhost:4318/v1/traces"`
|
||||
SendTelemetry bool `env:"MG_SEND_TELEMETRY" envDefault:"true"`
|
||||
InstanceID string `env:"MG_TIMESCALE_WRITER_INSTANCE_ID" envDefault:""`
|
||||
TraceRatio float64 `env:"MG_JAEGER_TRACE_RATIO" envDefault:"1.0"`
|
||||
LogLevel string `env:"MG_TIMESCALE_WRITER_LOG_LEVEL" envDefault:"info"`
|
||||
ConfigPath string `env:"MG_TIMESCALE_WRITER_CONFIG_PATH" envDefault:"/config.toml"`
|
||||
BrokerURL string `env:"MG_MESSAGE_BROKER_URL" envDefault:"nats://localhost:4222"`
|
||||
BrokerClientCert string `env:"MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT" envDefault:""`
|
||||
BrokerClientKey string `env:"MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY" envDefault:""`
|
||||
BrokerCACerts string `env:"MG_TIMESCALE_WRITER_BROKER_CA_CERTS" envDefault:""`
|
||||
JaegerURL url.URL `env:"MG_JAEGER_URL" envDefault:"http://localhost:4318/v1/traces"`
|
||||
SendTelemetry bool `env:"MG_SEND_TELEMETRY" envDefault:"true"`
|
||||
InstanceID string `env:"MG_TIMESCALE_WRITER_INSTANCE_ID" envDefault:""`
|
||||
TraceRatio float64 `env:"MG_JAEGER_TRACE_RATIO" envDefault:"1.0"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -112,7 +116,16 @@ func main() {
|
||||
repo := newService(db, logger)
|
||||
repo = consumertracing.NewBlocking(tracer, repo, httpServerConfig)
|
||||
|
||||
pubSub, err := brokers.NewPubSub(ctx, cfg.BrokerURL, logger)
|
||||
// The mTLS client identity is optional: brokers that expose an unauthenticated
|
||||
// listener need none. Passing the option with empty paths would fail the
|
||||
// connection outright, so only ask for it once one of them is configured. A
|
||||
// partial configuration still fails, since half an identity is not a usable one.
|
||||
var brokerOpts []messaging.Option
|
||||
if cfg.BrokerClientCert != "" || cfg.BrokerClientKey != "" || cfg.BrokerCACerts != "" {
|
||||
brokerOpts = append(brokerOpts, writers.InternalMetadata(cfg.BrokerClientCert, cfg.BrokerClientKey, cfg.BrokerCACerts))
|
||||
}
|
||||
|
||||
pubSub, err := writers.NewPubSub(ctx, cfg.BrokerURL, logger, brokerOpts...)
|
||||
if err != nil {
|
||||
logger.Error(fmt.Sprintf("failed to connect to message broker: %s", err))
|
||||
exitCode = 1
|
||||
@@ -121,7 +134,7 @@ func main() {
|
||||
defer pubSub.Close()
|
||||
pubSub = brokerstracing.NewPubSub(httpServerConfig, tracer, pubSub)
|
||||
|
||||
if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, brokers.AllTopic, logger); err != nil {
|
||||
if err = consumers.Start(ctx, svcName, pubSub, repo, cfg.ConfigPath, writers.AllTopic, logger); err != nil {
|
||||
logger.Error(fmt.Sprintf("failed to create Timescale writer: %s", err))
|
||||
exitCode = 1
|
||||
return
|
||||
|
||||
+16
@@ -502,6 +502,14 @@ MG_POSTGRES_WRITER_HTTP_PORT=9007
|
||||
MG_POSTGRES_WRITER_HTTP_SERVER_CERT=
|
||||
MG_POSTGRES_WRITER_HTTP_SERVER_KEY=
|
||||
MG_POSTGRES_WRITER_INSTANCE_ID=
|
||||
# The writer is a passive consumer on FluxMQ's mTLS local listener. Its client
|
||||
# certificate identifies the local principal and its SASL secret must match
|
||||
# docker/fluxmq/secrets/postgres-writer-current.
|
||||
MG_POSTGRES_WRITER_BROKER_SECRET=postgres-writer-fluxmq-local-principal-secret-change-me
|
||||
MG_POSTGRES_WRITER_BROKER_URL=amqps://postgres-writer:${MG_POSTGRES_WRITER_BROKER_SECRET}@fluxmq-node1:5685/
|
||||
MG_POSTGRES_WRITER_BROKER_CLIENT_CERT=/etc/ssl/certs/postgres-writer-fluxmq-client.crt
|
||||
MG_POSTGRES_WRITER_BROKER_CLIENT_KEY=/etc/ssl/private/postgres-writer-fluxmq-client.key
|
||||
MG_POSTGRES_WRITER_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
|
||||
|
||||
### Postgres Reader
|
||||
MG_POSTGRES_READER_LOG_LEVEL=debug
|
||||
@@ -529,6 +537,14 @@ MG_TIMESCALE_WRITER_HTTP_PORT=9012
|
||||
MG_TIMESCALE_WRITER_HTTP_SERVER_CERT=
|
||||
MG_TIMESCALE_WRITER_HTTP_SERVER_KEY=
|
||||
MG_TIMESCALE_WRITER_INSTANCE_ID=
|
||||
# The writer is a passive consumer on FluxMQ's mTLS local listener. Its client
|
||||
# certificate identifies the local principal and its SASL secret must match
|
||||
# docker/fluxmq/secrets/timescale-writer-current.
|
||||
MG_TIMESCALE_WRITER_BROKER_SECRET=timescale-writer-fluxmq-local-principal-secret-change-me
|
||||
MG_TIMESCALE_WRITER_BROKER_URL=amqps://timescale-writer:${MG_TIMESCALE_WRITER_BROKER_SECRET}@fluxmq-node1:5685/
|
||||
MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT=/etc/ssl/certs/timescale-writer-fluxmq-client.crt
|
||||
MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY=/etc/ssl/private/timescale-writer-fluxmq-client.key
|
||||
MG_TIMESCALE_WRITER_BROKER_CA_CERTS=/etc/ssl/certs/ca.crt
|
||||
|
||||
### Timescale Reader
|
||||
MG_TIMESCALE_READER_LOG_LEVEL=debug
|
||||
|
||||
@@ -35,7 +35,10 @@ services:
|
||||
image: ghcr.io/absmach/magistrala/postgres-writer:${MG_RELEASE_TAG}
|
||||
container_name: magistrala-postgres-writer
|
||||
depends_on:
|
||||
- postgres
|
||||
postgres:
|
||||
condition: service_started
|
||||
fluxmq-node1:
|
||||
condition: service_healthy
|
||||
restart: on-failure
|
||||
environment:
|
||||
MG_POSTGRES_WRITER_LOG_LEVEL: ${MG_POSTGRES_WRITER_LOG_LEVEL}
|
||||
@@ -53,7 +56,10 @@ services:
|
||||
MG_POSTGRES_SSL_CERT: ${MG_POSTGRES_SSL_CERT}
|
||||
MG_POSTGRES_SSL_KEY: ${MG_POSTGRES_SSL_KEY}
|
||||
MG_POSTGRES_SSL_ROOT_CERT: ${MG_POSTGRES_SSL_ROOT_CERT}
|
||||
MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
|
||||
MG_MESSAGE_BROKER_URL: ${MG_POSTGRES_WRITER_BROKER_URL}
|
||||
MG_POSTGRES_WRITER_BROKER_CLIENT_CERT: ${MG_POSTGRES_WRITER_BROKER_CLIENT_CERT}
|
||||
MG_POSTGRES_WRITER_BROKER_CLIENT_KEY: ${MG_POSTGRES_WRITER_BROKER_CLIENT_KEY}
|
||||
MG_POSTGRES_WRITER_BROKER_CA_CERTS: ${MG_POSTGRES_WRITER_BROKER_CA_CERTS}
|
||||
MG_JAEGER_URL: ${MG_JAEGER_URL}
|
||||
MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
|
||||
MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
|
||||
@@ -63,4 +69,11 @@ services:
|
||||
networks:
|
||||
- magistrala-base-net
|
||||
volumes:
|
||||
- ./config.toml:/config.toml
|
||||
# Relative paths in an addon file resolve against the project directory,
|
||||
# which is the base compose file's directory rather than this one.
|
||||
- ./addons/postgres-writer/config.toml:${MG_POSTGRES_WRITER_CONFIG_PATH}
|
||||
# The URI SAN in this client certificate identifies the Postgres writer
|
||||
# to the FluxMQ local listener; the broker ACL only permits `writers`.
|
||||
- ./ssl/certs/postgres-writer-fluxmq-client.crt:${MG_POSTGRES_WRITER_BROKER_CLIENT_CERT}:ro
|
||||
- ./ssl/certs/postgres-writer-fluxmq-client.key:${MG_POSTGRES_WRITER_BROKER_CLIENT_KEY}:ro
|
||||
- ./ssl/certs/ca.crt:${MG_POSTGRES_WRITER_BROKER_CA_CERTS}:ro
|
||||
|
||||
+72
-26
@@ -255,12 +255,14 @@ services:
|
||||
ipv4_address: 172.30.0.201
|
||||
volumes:
|
||||
- ./fluxmq/node1.yaml:/etc/fluxmq/config.yaml:ro
|
||||
# Server certificate and CA for the mTLS service listener, plus the
|
||||
# local-principal secret it authenticates the Rules Engine against.
|
||||
# Server certificate and CA for the mTLS service listener, plus each
|
||||
# local principal's SASL secret.
|
||||
- ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
|
||||
- ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
|
||||
- ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
|
||||
- ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
|
||||
- ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
|
||||
- ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
|
||||
- magistrala-fluxmq-node1-volume:/tmp/fluxmq
|
||||
|
||||
fluxmq-node2:
|
||||
@@ -281,12 +283,14 @@ services:
|
||||
ipv4_address: 172.30.0.202
|
||||
volumes:
|
||||
- ./fluxmq/node2.yaml:/etc/fluxmq/config.yaml:ro
|
||||
# Server certificate and CA for the mTLS service listener, plus the
|
||||
# local-principal secret it authenticates the Rules Engine against.
|
||||
# Server certificate and CA for the mTLS service listener, plus each
|
||||
# local principal's SASL secret.
|
||||
- ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
|
||||
- ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
|
||||
- ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
|
||||
- ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
|
||||
- ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
|
||||
- ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
|
||||
- magistrala-fluxmq-node2-volume:/tmp/fluxmq
|
||||
|
||||
fluxmq-node3:
|
||||
@@ -307,12 +311,14 @@ services:
|
||||
ipv4_address: 172.30.0.203
|
||||
volumes:
|
||||
- ./fluxmq/node3.yaml:/etc/fluxmq/config.yaml:ro
|
||||
# Server certificate and CA for the mTLS service listener, plus the
|
||||
# local-principal secret it authenticates the Rules Engine against.
|
||||
# Server certificate and CA for the mTLS service listener, plus each
|
||||
# local principal's SASL secret.
|
||||
- ./ssl/certs/fluxmq-service-server.crt:/etc/fluxmq/certs/fluxmq-service-server.crt:ro
|
||||
- ./ssl/certs/fluxmq-service-server.key:/etc/fluxmq/certs/fluxmq-service-server.key:ro
|
||||
- ./ssl/certs/ca.crt:/etc/fluxmq/certs/ca.crt:ro
|
||||
- ./fluxmq/secrets/re-current:/etc/fluxmq/secrets/re-current:ro
|
||||
- ./fluxmq/secrets/timescale-writer-current:/etc/fluxmq/secrets/timescale-writer-current:ro
|
||||
- ./fluxmq/secrets/postgres-writer-current:/etc/fluxmq/secrets/postgres-writer-current:ro
|
||||
- magistrala-fluxmq-node3-volume:/tmp/fluxmq
|
||||
|
||||
fluxmq-auth:
|
||||
@@ -557,8 +563,12 @@ services:
|
||||
networks:
|
||||
- magistrala-base-net
|
||||
|
||||
# Creates the image bucket with the broker's own image rather than pulling a
|
||||
# second one. An aws-cli here cost a 130 MB image for a single call, and it
|
||||
# was the only untagged reference in this file, so what it resolved to
|
||||
# depended on the day it was pulled.
|
||||
seaweedfs-init:
|
||||
image: amazon/aws-cli
|
||||
image: chrislusf/seaweedfs:4.40
|
||||
container_name: magistrala-seaweedfs-init
|
||||
entrypoint: /bin/sh
|
||||
depends_on:
|
||||
@@ -566,26 +576,46 @@ services:
|
||||
command:
|
||||
- -c
|
||||
- |
|
||||
echo "[INIT] Waiting 20s for SeaweedFS S3 to be ready...";
|
||||
sleep 20;
|
||||
OUT=$(aws --endpoint-url http://seaweedfs-s3:8333 s3api create-bucket --bucket $${BUCKET} 2>&1);
|
||||
EXIT=$$?;
|
||||
if [ $$EXIT -eq 0 ]; then
|
||||
echo "[INIT] Bucket $${BUCKET} created successfully.";
|
||||
elif echo "$$OUT" | grep -q 'BucketAlreadyOwnedByYou\|BucketAlreadyExists'; then
|
||||
echo "[INIT] Bucket $${BUCKET} already exists, skipping.";
|
||||
else
|
||||
echo "[INIT] Failed to create bucket $${BUCKET}: $$OUT" >&2;
|
||||
exit 1;
|
||||
fi
|
||||
set -eu
|
||||
# weed shell blocks indefinitely against an unreachable master, so the
|
||||
# wait is a port check and every weed call is bounded; left to itself it
|
||||
# would hang this container rather than fail it.
|
||||
#
|
||||
# 60 ticks is ~2 minutes while the store is merely slow to boot, and
|
||||
# ~3 minutes to give up when it never appears, since a failing lookup
|
||||
# costs about a second on top of the sleep.
|
||||
tick=0;
|
||||
while ! nc -z seaweedfs-s3 9333 2>/dev/null; do
|
||||
tick=$$((tick + 1));
|
||||
if [ $$tick -ge 60 ]; then
|
||||
echo "[INIT] SeaweedFS master did not come up" >&2;
|
||||
exit 1;
|
||||
fi;
|
||||
sleep 2;
|
||||
done;
|
||||
printf 's3.bucket.create -name %s\n' "$${BUCKET}" > /tmp/create.cmd;
|
||||
printf 's3.bucket.list\n' > /tmp/list.cmd;
|
||||
try=0;
|
||||
while [ $$try -lt 3 ]; do
|
||||
try=$$((try + 1));
|
||||
timeout 20 weed shell -master seaweedfs-s3:9333 < /tmp/create.cmd > /tmp/create.out 2>&1 || true;
|
||||
timeout 20 weed shell -master seaweedfs-s3:9333 < /tmp/list.cmd > /tmp/list.out 2>/dev/null || true;
|
||||
# s3.bucket.create reports success even for an existing bucket, and
|
||||
# weed shell exits 0 whatever the command did, so the listing is the
|
||||
# only trustworthy signal that the bucket is really there.
|
||||
if awk -v b="$${BUCKET}" '{sub(/^[ \t]+/,""); split($$0,f,/[ \t]/); if (f[1]==b) found=1} END{exit !found}' /tmp/list.out; then
|
||||
echo "[INIT] Bucket $${BUCKET} present.";
|
||||
exit 0;
|
||||
fi;
|
||||
sleep 3;
|
||||
done;
|
||||
echo "[INIT] Bucket $${BUCKET} was not created after $$try attempts" >&2;
|
||||
cat /tmp/create.out >&2;
|
||||
exit 1
|
||||
networks:
|
||||
- magistrala-base-net
|
||||
environment:
|
||||
BUCKET: ${MG_BACKEND_OBJECT_STORAGE_BUCKET}
|
||||
AWS_ACCESS_KEY_ID: ${MG_BACKEND_OBJECT_STORAGE_ACCESS_KEY}
|
||||
AWS_SECRET_ACCESS_KEY: ${MG_BACKEND_OBJECT_STORAGE_SECRET_KEY}
|
||||
AWS_DEFAULT_REGION: ${MG_BACKEND_OBJECT_STORAGE_REGION}
|
||||
AWS_EC2_METADATA_DISABLED: "true"
|
||||
|
||||
timescale:
|
||||
image: timescale/timescaledb:2.29.0-pg16-oss
|
||||
@@ -601,13 +631,18 @@ services:
|
||||
- magistrala-base-net
|
||||
volumes:
|
||||
- magistrala-timescale-writer-volume:/var/lib/postgresql/data
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
|
||||
interval: 5s
|
||||
timeout: 3s
|
||||
retries: 60
|
||||
|
||||
timescale-reader:
|
||||
image: ghcr.io/absmach/magistrala/timescale-reader:${MG_RELEASE_TAG}
|
||||
container_name: magistrala-timescale-reader
|
||||
depends_on:
|
||||
timescale:
|
||||
condition: service_started
|
||||
condition: service_healthy
|
||||
atom-bootstrap:
|
||||
condition: service_completed_successfully
|
||||
restart: on-failure
|
||||
@@ -686,7 +721,10 @@ services:
|
||||
image: ghcr.io/absmach/magistrala/timescale-writer:${MG_RELEASE_TAG}
|
||||
container_name: magistrala-timescale-writer
|
||||
depends_on:
|
||||
- timescale
|
||||
timescale:
|
||||
condition: service_healthy
|
||||
fluxmq-node1:
|
||||
condition: service_healthy
|
||||
restart: on-failure
|
||||
environment:
|
||||
MG_TIMESCALE_WRITER_LOG_LEVEL: ${MG_TIMESCALE_WRITER_LOG_LEVEL}
|
||||
@@ -704,7 +742,10 @@ services:
|
||||
MG_TIMESCALE_SSL_CERT: ${MG_TIMESCALE_SSL_CERT}
|
||||
MG_TIMESCALE_SSL_KEY: ${MG_TIMESCALE_SSL_KEY}
|
||||
MG_TIMESCALE_SSL_ROOT_CERT: ${MG_TIMESCALE_SSL_ROOT_CERT}
|
||||
MG_MESSAGE_BROKER_URL: ${MG_MESSAGE_BROKER_URL}
|
||||
MG_MESSAGE_BROKER_URL: ${MG_TIMESCALE_WRITER_BROKER_URL}
|
||||
MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT: ${MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT}
|
||||
MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY: ${MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY}
|
||||
MG_TIMESCALE_WRITER_BROKER_CA_CERTS: ${MG_TIMESCALE_WRITER_BROKER_CA_CERTS}
|
||||
MG_JAEGER_URL: ${MG_JAEGER_URL}
|
||||
MG_JAEGER_TRACE_RATIO: ${MG_JAEGER_TRACE_RATIO}
|
||||
MG_SEND_TELEMETRY: ${MG_SEND_TELEMETRY}
|
||||
@@ -715,6 +756,11 @@ services:
|
||||
- magistrala-base-net
|
||||
volumes:
|
||||
- ./addons/timescale-writer/config.toml:${MG_TIMESCALE_WRITER_CONFIG_PATH}
|
||||
# The URI SAN in this client certificate identifies the Timescale writer
|
||||
# to the FluxMQ local listener; the broker ACL only permits `writers`.
|
||||
- ./ssl/certs/timescale-writer-fluxmq-client.crt:${MG_TIMESCALE_WRITER_BROKER_CLIENT_CERT}:ro
|
||||
- ./ssl/certs/timescale-writer-fluxmq-client.key:${MG_TIMESCALE_WRITER_BROKER_CLIENT_KEY}:ro
|
||||
- ./ssl/certs/ca.crt:${MG_TIMESCALE_WRITER_BROKER_CA_CERTS}:ro
|
||||
re-db:
|
||||
image: docker.io/postgres:18.4-alpine3.24
|
||||
container_name: magistrala-re-db
|
||||
|
||||
@@ -143,7 +143,7 @@ queues:
|
||||
auth:
|
||||
# Principals admitted on the service listener. The certificate_uri_san must
|
||||
# match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
|
||||
# the secret file must hold the same value as MG_RE_BROKER_SECRET.
|
||||
# each secret file must hold the value configured by its service.
|
||||
local_principals:
|
||||
- name: "rules-engine"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
|
||||
@@ -164,6 +164,24 @@ auth:
|
||||
# The message stream the Rules Engine consumes.
|
||||
subscribe:
|
||||
- "m"
|
||||
- name: "timescale-writer"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
|
||||
role: "service"
|
||||
current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
|
||||
permissions:
|
||||
# The writer is a passive consumer of the broker-provisioned stream. It
|
||||
# neither publishes messages nor manages queues.
|
||||
subscribe:
|
||||
- "writers"
|
||||
- name: "postgres-writer"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
|
||||
role: "service"
|
||||
current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
|
||||
permissions:
|
||||
# The writer is a passive consumer of the broker-provisioned stream. It
|
||||
# neither publishes messages nor manages queues.
|
||||
subscribe:
|
||||
- "writers"
|
||||
external:
|
||||
url: "http://fluxmq-auth:7016"
|
||||
transport: "grpc"
|
||||
|
||||
@@ -140,7 +140,7 @@ queues:
|
||||
auth:
|
||||
# Principals admitted on the service listener. The certificate_uri_san must
|
||||
# match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
|
||||
# the secret file must hold the same value as MG_RE_BROKER_SECRET.
|
||||
# each secret file must hold the value configured by its service.
|
||||
local_principals:
|
||||
- name: "rules-engine"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
|
||||
@@ -161,6 +161,24 @@ auth:
|
||||
# The message stream the Rules Engine consumes.
|
||||
subscribe:
|
||||
- "m"
|
||||
- name: "timescale-writer"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
|
||||
role: "service"
|
||||
current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
|
||||
permissions:
|
||||
# The writer is a passive consumer of the broker-provisioned stream. It
|
||||
# neither publishes messages nor manages queues.
|
||||
subscribe:
|
||||
- "writers"
|
||||
- name: "postgres-writer"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
|
||||
role: "service"
|
||||
current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
|
||||
permissions:
|
||||
# The writer is a passive consumer of the broker-provisioned stream. It
|
||||
# neither publishes messages nor manages queues.
|
||||
subscribe:
|
||||
- "writers"
|
||||
external:
|
||||
url: "http://fluxmq-auth:7016"
|
||||
transport: "grpc"
|
||||
|
||||
@@ -140,7 +140,7 @@ queues:
|
||||
auth:
|
||||
# Principals admitted on the service listener. The certificate_uri_san must
|
||||
# match the URI SAN issued by docker/ssl/Makefile (fluxmq_service_certs), and
|
||||
# the secret file must hold the same value as MG_RE_BROKER_SECRET.
|
||||
# each secret file must hold the value configured by its service.
|
||||
local_principals:
|
||||
- name: "rules-engine"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/rules-engine"
|
||||
@@ -161,6 +161,24 @@ auth:
|
||||
# The message stream the Rules Engine consumes.
|
||||
subscribe:
|
||||
- "m"
|
||||
- name: "timescale-writer"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/timescale-writer"
|
||||
role: "service"
|
||||
current_secret_file: "/etc/fluxmq/secrets/timescale-writer-current"
|
||||
permissions:
|
||||
# The writer is a passive consumer of the broker-provisioned stream. It
|
||||
# neither publishes messages nor manages queues.
|
||||
subscribe:
|
||||
- "writers"
|
||||
- name: "postgres-writer"
|
||||
certificate_uri_san: "spiffe://absmach/magistrala/postgres-writer"
|
||||
role: "service"
|
||||
current_secret_file: "/etc/fluxmq/secrets/postgres-writer-current"
|
||||
permissions:
|
||||
# The writer is a passive consumer of the broker-provisioned stream. It
|
||||
# neither publishes messages nor manages queues.
|
||||
subscribe:
|
||||
- "writers"
|
||||
external:
|
||||
url: "http://fluxmq-auth:7016"
|
||||
transport: "grpc"
|
||||
|
||||
+42
-14
@@ -51,6 +51,12 @@ FLUXMQ_SERVICE_SERVER_CRT_FILE_NAME=fluxmq-service-server
|
||||
RE_FLUXMQ_CLIENT_CN=rules-engine
|
||||
RE_FLUXMQ_CLIENT_CRT_FILE_NAME=re-fluxmq-client
|
||||
RE_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/rules-engine
|
||||
TIMESCALE_WRITER_FLUXMQ_CLIENT_CN=timescale-writer
|
||||
TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME=timescale-writer-fluxmq-client
|
||||
TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/timescale-writer
|
||||
POSTGRES_WRITER_FLUXMQ_CLIENT_CN=postgres-writer
|
||||
POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME=postgres-writer-fluxmq-client
|
||||
POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN=spiffe://absmach/magistrala/postgres-writer
|
||||
|
||||
define GRPC_CERT_CONFIG
|
||||
[req]
|
||||
@@ -131,7 +137,7 @@ It can be downloaded from $(DOWNLOAD_URL).
|
||||
|
||||
etc, etc.
|
||||
endef
|
||||
all: clean_certs ca server_cert auth_grpc_certs domains_grpc_certs groups_grpc_certs clients_grpc_certs channels_grpc_certs coap_dtls_certs fluxmq_service_certs fluxmq_service_secret re_trace_key
|
||||
all: clean_certs ca server_cert auth_grpc_certs domains_grpc_certs groups_grpc_certs clients_grpc_certs channels_grpc_certs coap_dtls_certs fluxmq_service_certs fluxmq_service_secret timescale_writer_fluxmq_service_secret postgres_writer_fluxmq_service_secret re_trace_key
|
||||
|
||||
# CA name and key is "ca".
|
||||
ca:
|
||||
@@ -245,31 +251,53 @@ define gen_uri_san_cert
|
||||
endef
|
||||
|
||||
# Certificates for the FluxMQ service listener: a server certificate for the
|
||||
# broker and a client certificate carrying the Rules Engine's URI SAN. The
|
||||
# broker and a client certificate carrying each local principal's URI SAN. The
|
||||
# listener requires mTLS, so both sides are issued by the same CA.
|
||||
fluxmq_service_certs:
|
||||
$(call gen_fluxmq_server_cert,$(FLUXMQ_SERVICE_SERVER_CRT_FILE_NAME))
|
||||
$(call gen_uri_san_cert,$(RE_FLUXMQ_CLIENT_CRT_FILE_NAME),$(RE_FLUXMQ_CLIENT_CN),$(RE_FLUXMQ_CLIENT_URI_SAN))
|
||||
$(call gen_uri_san_cert,$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CN),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN))
|
||||
$(call gen_uri_san_cert,$(POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(POSTGRES_WRITER_FLUXMQ_CLIENT_CN),$(POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN))
|
||||
|
||||
# The SASL secret the Rules Engine presents alongside its certificate. FluxMQ
|
||||
# reads it from a file and the Rules Engine sends it in its broker URL, so both
|
||||
# sides must see the same value: .env is the single source and this target
|
||||
# derives the file from it. FluxMQ requires at least 32 bytes.
|
||||
fluxmq_service_secret:
|
||||
# Generate a single principal's certificate when adding it to an existing
|
||||
# deployment, without rotating the broker certificate or the other principals.
|
||||
timescale_writer_fluxmq_client_cert:
|
||||
$(call gen_uri_san_cert,$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_CN),$(TIMESCALE_WRITER_FLUXMQ_CLIENT_URI_SAN))
|
||||
|
||||
postgres_writer_fluxmq_client_cert:
|
||||
$(call gen_uri_san_cert,$(POSTGRES_WRITER_FLUXMQ_CLIENT_CRT_FILE_NAME),$(POSTGRES_WRITER_FLUXMQ_CLIENT_CN),$(POSTGRES_WRITER_FLUXMQ_CLIENT_URI_SAN))
|
||||
|
||||
# The SASL secret a local principal presents alongside its certificate. FluxMQ
|
||||
# reads it from a file and the service sends it in its broker URL, so both sides
|
||||
# must see the same value: .env is the single source and these targets derive
|
||||
# the file from it. FluxMQ requires at least 32 bytes.
|
||||
#
|
||||
# Usage: $(call gen_fluxmq_service_secret,env_variable_name,secret_file_name)
|
||||
define gen_fluxmq_service_secret
|
||||
@mkdir -p ../fluxmq/secrets
|
||||
@secret="$${MG_RE_BROKER_SECRET}"; \
|
||||
@secret="$${$(1)}"; \
|
||||
if [ -z "$$secret" ]; then \
|
||||
secret=$$(grep -E "^MG_RE_BROKER_SECRET=" ../.env | cut -d= -f2-); \
|
||||
secret=$$(grep -E "^$(1)=" ../.env | cut -d= -f2-); \
|
||||
fi; \
|
||||
if [ -z "$$secret" ]; then \
|
||||
echo "MG_RE_BROKER_SECRET is not set in docker/.env"; exit 1; \
|
||||
echo "$(1) is not set in docker/.env"; exit 1; \
|
||||
fi; \
|
||||
if [ $${#secret} -lt 32 ]; then \
|
||||
echo "MG_RE_BROKER_SECRET must be at least 32 characters"; exit 1; \
|
||||
echo "$(1) must be at least 32 characters"; exit 1; \
|
||||
fi; \
|
||||
printf '%s' "$$secret" > ../fluxmq/secrets/re-current; \
|
||||
chmod 600 ../fluxmq/secrets/re-current; \
|
||||
echo "wrote docker/fluxmq/secrets/re-current"
|
||||
printf '%s' "$$secret" > ../fluxmq/secrets/$(2); \
|
||||
chmod 600 ../fluxmq/secrets/$(2); \
|
||||
echo "wrote docker/fluxmq/secrets/$(2)"
|
||||
endef
|
||||
|
||||
fluxmq_service_secret:
|
||||
$(call gen_fluxmq_service_secret,MG_RE_BROKER_SECRET,re-current)
|
||||
|
||||
timescale_writer_fluxmq_service_secret:
|
||||
$(call gen_fluxmq_service_secret,MG_TIMESCALE_WRITER_BROKER_SECRET,timescale-writer-current)
|
||||
|
||||
postgres_writer_fluxmq_service_secret:
|
||||
$(call gen_fluxmq_service_secret,MG_POSTGRES_WRITER_BROKER_SECRET,postgres-writer-current)
|
||||
|
||||
# Persistent HMAC key shared by Rules Engine replicas. Do not replace an
|
||||
# existing key: doing so while old messages are in flight would invalidate
|
||||
|
||||
@@ -7,7 +7,7 @@ require (
|
||||
connectrpc.com/otelconnect v0.9.0
|
||||
github.com/0x6flab/namegenerator v1.4.0
|
||||
github.com/absmach/callhome v0.18.2
|
||||
github.com/absmach/fluxmq v0.50.0
|
||||
github.com/absmach/fluxmq v0.50.1-0.20260804171048-485d54e12a62
|
||||
github.com/absmach/senml v1.0.8
|
||||
github.com/caarlos0/env/v10 v10.0.0
|
||||
github.com/caarlos0/env/v11 v11.4.1
|
||||
|
||||
@@ -20,8 +20,8 @@ github.com/VividCortex/gohistogram v1.0.0 h1:6+hBz+qvs0JOrrNhhmR7lFxo5sINxBCGXrd
|
||||
github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=
|
||||
github.com/absmach/callhome v0.18.2 h1:dmopRHm2qTheHN1hdUKRRYpKwRrj7X9d8AWCFrb+K6s=
|
||||
github.com/absmach/callhome v0.18.2/go.mod h1:LEXKhES9JJtj3tBgTZv7VPNjOi5ukJQB0mFic0QP60Q=
|
||||
github.com/absmach/fluxmq v0.50.0 h1:RlYRv285ONSG6l0VqnFin7RLF8ZwAG2dkuHr6yn8iNo=
|
||||
github.com/absmach/fluxmq v0.50.0/go.mod h1:7sTd2lkxcq915f3Zp7mj1wXCBsuw9tdW+GT7GsLs8SI=
|
||||
github.com/absmach/fluxmq v0.50.1-0.20260804171048-485d54e12a62 h1:RhuJJ9vO3+6G2LzSqAKXaHBLRzhrE/C4tNq967AKWtw=
|
||||
github.com/absmach/fluxmq v0.50.1-0.20260804171048-485d54e12a62/go.mod h1:7sTd2lkxcq915f3Zp7mj1wXCBsuw9tdW+GT7GsLs8SI=
|
||||
github.com/absmach/senml v1.0.8 h1:+opem/r4g6c6eA/JLyCIuksyEhj7eBdysY3pEmy1mqo=
|
||||
github.com/absmach/senml v1.0.8/go.mod h1:DRhzHLgvQoIUHroBgpFrSWso+bJZO9E96RlHAHy+VRI=
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
|
||||
@@ -34,8 +34,23 @@ var cfg = jetstream.StreamConfig{
|
||||
Storage: jetstream.FileStorage,
|
||||
}
|
||||
|
||||
func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.PubSub, error) {
|
||||
pb, err := broker.NewPubSub(ctx, url, logger, broker.Prefix(prefix), broker.JSStreamConfig(cfg), broker.ConnectionName("writers-msg-pubsub"))
|
||||
// InternalMetadata returns an option for a trusted local-service connection
|
||||
// that carries internal metadata over mTLS and consumes the broker-provisioned
|
||||
// writers stream. It lives here rather than being taken from
|
||||
// pkg/messaging/brokers because that package selects its backend on a different
|
||||
// build tag: an untagged build would pair a FluxMQ option with a NATS PubSub.
|
||||
func InternalMetadata(certFile, keyFile, caFile string) messaging.Option {
|
||||
return broker.InternalMetadata(certFile, keyFile, caFile)
|
||||
}
|
||||
|
||||
func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...messaging.Option) (messaging.PubSub, error) {
|
||||
brokerOpts := []messaging.Option{
|
||||
broker.Prefix(prefix),
|
||||
broker.JSStreamConfig(cfg),
|
||||
broker.ConnectionName("writers-msg-pubsub"),
|
||||
}
|
||||
brokerOpts = append(brokerOpts, opts...)
|
||||
pb, err := broker.NewPubSub(ctx, url, logger, brokerOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -34,8 +34,20 @@ var cfg = jetstream.StreamConfig{
|
||||
Storage: jetstream.FileStorage,
|
||||
}
|
||||
|
||||
func NewPubSub(ctx context.Context, url string, logger *slog.Logger) (messaging.PubSub, error) {
|
||||
pb, err := broker.NewPubSub(ctx, url, logger, broker.Prefix(prefix), broker.JSStreamConfig(cfg))
|
||||
// InternalMetadata is a no-op for the NATS backend. It exists for compile-time
|
||||
// compatibility with the FluxMQ variant; NATS carries metadata in the protobuf
|
||||
// message.
|
||||
func InternalMetadata(_, _, _ string) messaging.Option {
|
||||
return func(_ any) error { return nil }
|
||||
}
|
||||
|
||||
func NewPubSub(ctx context.Context, url string, logger *slog.Logger, opts ...messaging.Option) (messaging.PubSub, error) {
|
||||
brokerOpts := []messaging.Option{
|
||||
broker.Prefix(prefix),
|
||||
broker.JSStreamConfig(cfg),
|
||||
}
|
||||
brokerOpts = append(brokerOpts, opts...)
|
||||
pb, err := broker.NewPubSub(ctx, url, logger, brokerOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user