pkg/events: Add support for a global CloudAMQP rabbit setup (#80)

* pkg/event: Set server name in TLS config

* docker: Install CA certificates

* Revert "pkg/event: Set server name in TLS config"

This reverts commit f651a11.

* Dockerfile: a little more strict on alpine

* Revert "Revert "pkg/event: Set server name in TLS config""

This reverts commit d5b6959.

* pkg/event: Remove port from server certificate check

* pkg/event: Remove port fallback logic

wtf

* reducers: Allow changing stream state exchange

* cmd/analyzer: Make the default stream state exchange mist

* pkg/event: Set tls stream name on healthcheck as well

* health: Stop checking connection on healthcheck

not scaling with cloud rabbits

* reducers: Make stream state non optional

Dockerfile

@@ -18,7 +18,9 @@ COPY	.	.
 
 RUN	make "version=$version"
 
-FROM	alpine:latest
+FROM	alpine:3.16
+
+RUN	apk add --no-cache ca-certificates
 
 WORKDIR	/app
 

cmd/analyzer/analyzer.go

@@ -31,9 +31,10 @@ type cliFlags struct {
 	rabbitmqUri string
 	amqpUri     string
 
-	golivepeerExchange string
-	shardPrefixesFlag  string
-	shardPrefixes      []string
+	golivepeerExchange  string
+	shardPrefixesFlag   string
+	shardPrefixes       []string
+	streamStateExchange string
 
 	serverOpts       api.ServerOptions
 	streamingOpts    health.StreamingOptions
@@ -53,6 +54,7 @@ func parseFlags(version string) cliFlags {
 
 	fs.StringVar(&cli.golivepeerExchange, "golivepeer-exchange", "lp_golivepeer_metadata", "Name of RabbitMQ exchange to bind the stream to on creation")
 	fs.StringVar(&cli.shardPrefixesFlag, "shard-prefixes", "", "Comma-separated list of prefixes of manifest IDs to process events from")
+	fs.StringVar(&cli.streamStateExchange, "stream-state-exchange", "lp_mist_api_connector", "Name of RabbitMQ exchange where to receive stream state events")
 
 	// Server options
 	fs.StringVar(&cli.serverOpts.Host, "host", "localhost", "Hostname to bind to")
@@ -131,7 +133,7 @@ func Run(build BuildFlags) {
 	}
 	defer consumer.Close()
 
-	reducer := reducers.Default(cli.golivepeerExchange, cli.shardPrefixes)
+	reducer := reducers.Default(cli.golivepeerExchange, cli.shardPrefixes, cli.streamStateExchange)
 	healthcore := health.NewCore(health.CoreOptions{
 		Streaming:        cli.streamingOpts,
 		StartTimeOffset:  reducers.DefaultStarTimeOffset(),

health/core.go

@@ -88,11 +88,6 @@ func NewCore(opts CoreOptions, consumer event.StreamConsumer, reducer Reducer) *
 }
 
 func (c *Core) IsHealthy() bool {
-	err := c.consumer.CheckConnection()
-	if err != nil {
-		glog.Warningf("Health core is unhealthy. reason=consumerErr consumerErr=%q", err)
-		return false
-	}
 	if tol := c.opts.Streaming.EventFlowSilenceTolerance; tol > 0 && time.Since(c.lastEventTs) > tol {
 		glog.Warningf("Health core is unhealthy. reason=noEvents lastEventTs=%s, tolerance=%s", c.lastEventTs, tol)
 		return false

health/reducers/defaults.go

@@ -11,9 +11,9 @@ var (
 	maxStatsWindow = statsWindows[len(statsWindows)-1]
 )
 
-func Default(golpExchange string, shardPrefixes []string) health.Reducer {
+func Default(golpExchange string, shardPrefixes []string, streamStateExchange string) health.Reducer {
 	return Pipeline{
-		StreamStateReducer{},
+		StreamStateReducer{streamStateExchange},
 		TranscodeReducer{golpExchange, shardPrefixes},
 		MultistreamReducer{},
 		MediaServerMetrics{},

health/reducers/stream_state.go

@@ -9,7 +9,6 @@ import (
 )
 
 const (
-	globalExchange        = "lp_global_replication"
 	streamStateBindingKey = "stream.state.#"
 
 	ConditionActive data.ConditionType = "Active"
@@ -20,10 +19,12 @@ type ActiveConditionExtraData struct {
 	Region string `json:"region"`
 }
 
-type StreamStateReducer struct{}
+type StreamStateReducer struct {
+	exchange string
+}
 
 func (t StreamStateReducer) Bindings() []event.BindingArgs {
-	return []event.BindingArgs{{Exchange: globalExchange, Key: streamStateBindingKey}}
+	return []event.BindingArgs{{Exchange: t.exchange, Key: streamStateBindingKey}}
 }
 
 func (t StreamStateReducer) Conditions() []data.ConditionType {

pkg/event/stream_consumer.go

@@ -2,6 +2,7 @@ package event
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"net/url"
@@ -36,9 +37,7 @@ func NewStreamConsumer(streamUriStr, amqpUriStr string) (StreamConsumer, error)
 		return nil, err
 	}
 	glog.Infof("Connecting to RabbitMQ. streamUri=%q, amqpUri=%q", streamUri.Redacted(), amqpUri.Redacted())
-	opts := stream.NewEnvironmentOptions().
-		SetMaxConsumersPerClient(5).
-		SetUri(streamUri.String())
+	opts := baseEnvOpts(streamUri).SetMaxConsumersPerClient(5)
 	env, err := stream.NewEnvironment(opts)
 	if err != nil {
 		return nil, err
@@ -53,8 +52,7 @@ func (c *strmConsumer) Close() error {
 
 func (c *strmConsumer) CheckConnection() error {
 	// create separate env for test to avoid infinite connect retry from lib
-	env, err := stream.NewEnvironment(
-		stream.NewEnvironmentOptions().SetUri(c.streamUri.String()))
+	env, err := stream.NewEnvironment(baseEnvOpts(c.streamUri))
 	if err != nil {
 		return err
 	}
@@ -66,6 +64,14 @@ func (c *strmConsumer) CheckConnection() error {
 	return env.Close()
 }
 
+func baseEnvOpts(uri *url.URL) *stream.EnvironmentOptions {
+	return stream.NewEnvironmentOptions().
+		SetUri(uri.String()).
+		SetTLSConfig(&tls.Config{
+			ServerName: uri.Hostname(),
+		})
+}
+
 func (c *strmConsumer) ConsumeChan(ctx context.Context, opts ConsumeOptions) (<-chan StreamMessage, error) {
 	err := c.ensureStream(opts.Stream, opts.StreamOptions)
 	if err != nil {
@@ -267,9 +273,6 @@ func coalesceUri(value *url.URL, fallback url.URL) *url.URL {
 	if result.Scheme == "" {
 		result.Scheme = fallback.Scheme
 	}
-	if result.Port() == "" && fallback.Port() != "" {
-		result.Host += ":" + fallback.Port()
-	}
 	if u := result.User; u == nil || u.String() == "" {
 		result.User = fallback.User
 		if u := result.User; u == nil || u.String() == "" {