outbounder.go

// SPDX-FileCopyrightText: 2017 Comcast Cable Communications Management, LLC
// SPDX-License-Identifier: Apache-2.0
package main

import (
	"encoding/json"
	"net/http"
	"time"

	"github.com/spf13/viper"
	"github.com/xmidt-org/webpa-common/v2/adapter"
	"github.com/xmidt-org/webpa-common/v2/device"
	"github.com/xmidt-org/webpa-common/v2/event"
	"go.uber.org/zap"

	// nolint:staticcheck
	"github.com/xmidt-org/webpa-common/v2/xresolver"
	"github.com/xmidt-org/webpa-common/v2/xresolver/consul"
)

const (
	// OutbounderKey is the Viper subkey which is expected to hold Outbounder configuration
	OutbounderKey = "device.outbound"

	DefaultDefaultScheme = "https"
	DefaultAllowedScheme = "https"

	DefaultEventType                         = "default"
	DefaultMethod                            = "POST"
	DefaultRetries                           = 1
	DefaultWorkerPoolSize      uint          = 100
	DefaultSource              string        = "localhost"
	DefaultOutboundQueueSize   uint          = 1000
	DefaultRequestTimeout      time.Duration = 125 * time.Second
	DefaultClientTimeout       time.Duration = 160 * time.Second
	DefaultMaxIdleConns                      = 0
	DefaultMaxIdleConnsPerHost               = 100
	DefaultIdleConnTimeout     time.Duration = 0
)

// Outbounder encapsulates the configuration necessary for handling outbound traffic
// and grants the ability to start the outbounding infrastructure.
type Outbounder struct {
	Method                 string                 `json:"method"`
	Retries                int                    `json:"retries"`
	RequestTimeout         time.Duration          `json:"requestTimeout"`
	DefaultScheme          string                 `json:"defaultScheme"`
	AllowedSchemes         []string               `json:"allowedSchemes"`
	EventEndpoints         map[string]interface{} `json:"eventEndpoints"`
	EnableConsulRoundRobin bool                   `json:"enableConsulRoundRobin"`
	OutboundQueueSize      uint                   `json:"outboundQueueSize"`
	WorkerPoolSize         uint                   `json:"workerPoolSize"`
	Source                 string                 `json:"source"`
	Transport              http.Transport         `json:"transport"`
	ClientTimeout          time.Duration          `json:"clientTimeout"`
	AuthKey                string                 `json:"authKey"`
	Logger                 *zap.Logger            `json:"-"`
}

// NewOutbounder returns an Outbounder unmarshalled from a Viper environment.
// This function allows the Viper instance to be nil, in which case a default
// Outbounder is returned.
func NewOutbounder(logger *zap.Logger, v *viper.Viper) (o *Outbounder, watcher *consul.ConsulWatcher, err error) {
	options := consul.Options{
		Watch:  make(map[string]string),
		Logger: logger,
	}

	o = &Outbounder{
		Method:            DefaultMethod,
		RequestTimeout:    DefaultRequestTimeout,
		DefaultScheme:     DefaultDefaultScheme,
		AllowedSchemes:    []string{DefaultAllowedScheme},
		OutboundQueueSize: DefaultOutboundQueueSize,
		WorkerPoolSize:    DefaultWorkerPoolSize,
		Transport: http.Transport{
			MaxIdleConns:        DefaultMaxIdleConns,
			MaxIdleConnsPerHost: DefaultMaxIdleConnsPerHost,
			IdleConnTimeout:     DefaultIdleConnTimeout,
		},
		ClientTimeout: DefaultClientTimeout,
		Logger:        logger,
	}

	if v != nil {
		err = v.Unmarshal(o)
	}
	if o.EnableConsulRoundRobin {
		logger.Info("Using consul round robin on service discover", zap.String("service", "caduceus"))
		for _, url := range o.EventEndpoints {
			options.Watch[url.(string)] = "caduceus"
		}
		watcher = consul.NewConsulWatcher(options)
		o.Transport.DialContext = xresolver.NewResolver(xresolver.DefaultDialer, logger.With(zap.String("component", "xresolver")), watcher).DialContext
	}
	return
}

// String emits a JSON string representing this outbounder, primarily useful for debugging.
func (o *Outbounder) String() string {
	// nolint: staticcheck
	if data, err := json.Marshal(o); err != nil {
		return err.Error()
	} else {
		return string(data)
	}
}

func (o *Outbounder) logger() *zap.Logger {
	if o != nil && o.Logger != nil {
		return o.Logger
	}

	return adapter.DefaultLogger().Logger
}

func (o *Outbounder) method() string {
	if o != nil && len(o.Method) > 0 {
		return o.Method
	}

	return DefaultMethod
}

func (o *Outbounder) retries() int {
	if o != nil && o.Retries > 0 {
		return o.Retries
	}

	return DefaultRetries
}

func (o *Outbounder) requestTimeout() time.Duration {
	if o != nil && o.RequestTimeout > 0 {
		return o.RequestTimeout
	}

	return DefaultRequestTimeout
}

func (o *Outbounder) defaultScheme() string {
	if o != nil && len(o.DefaultScheme) > 0 {
		return o.DefaultScheme
	}

	return DefaultDefaultScheme
}

func (o *Outbounder) allowedSchemes() map[string]bool {
	if o != nil && len(o.AllowedSchemes) > 0 {
		allowedSchemes := make(map[string]bool, len(o.AllowedSchemes))
		for _, as := range o.AllowedSchemes {
			allowedSchemes[as] = true
		}

		return allowedSchemes
	}

	return map[string]bool{DefaultAllowedScheme: true}
}

func (o *Outbounder) eventMap() (event.MultiMap, error) {
	if o != nil {
		return event.NestedToMultiMap(".", o.EventEndpoints)
	}

	// don't return nil, as we want to make sure client code can
	// always do lookups for event types
	return event.MultiMap{}, nil
}

func (o *Outbounder) outboundQueueSize() uint {
	if o != nil && o.OutboundQueueSize > 0 {
		return o.OutboundQueueSize
	}

	return DefaultOutboundQueueSize
}

func (o *Outbounder) source() string {
	if o != nil && len(o.Source) > 0 {
		return o.Source
	}

	return DefaultSource
}

func (o *Outbounder) workerPoolSize() uint {
	if o != nil && o.WorkerPoolSize > 0 {
		return o.WorkerPoolSize
	}

	return DefaultWorkerPoolSize
}

func (o *Outbounder) transport() *http.Transport {
	if o != nil {
		copyOf := new(http.Transport)
		// nolint:govet
		*copyOf = o.Transport

		if copyOf.MaxIdleConns <= 0 {
			copyOf.MaxIdleConns = DefaultMaxIdleConns
		}

		if copyOf.MaxIdleConnsPerHost <= 0 {
			copyOf.MaxIdleConnsPerHost = DefaultMaxIdleConnsPerHost
		}

		if copyOf.IdleConnTimeout <= 0 {
			copyOf.IdleConnTimeout = DefaultIdleConnTimeout
		}

		return copyOf
	}

	return &http.Transport{
		MaxIdleConns:        DefaultMaxIdleConns,
		MaxIdleConnsPerHost: DefaultMaxIdleConnsPerHost,
		IdleConnTimeout:     DefaultIdleConnTimeout,
	}
}

func (o *Outbounder) authKey() string {
	if o != nil {
		return o.AuthKey
	}

	return ""
}

func (o *Outbounder) clientTimeout() time.Duration {
	if o != nil && o.ClientTimeout > 0 {
		return o.ClientTimeout
	}

	return DefaultClientTimeout
}

// Start spawns all necessary goroutines and returns a device.Listener
func (o *Outbounder) Start(om OutboundMeasures) ([]device.Listener, error) {
	logger := o.logger()
	logger.Info("Starting outbounder")
	eventDispatcher, outbounds, err := NewEventDispatcher(om, o, nil)
	if err != nil {
		return nil, err
	}

	workerPool := NewWorkerPool(om, o, outbounds)
	workerPool.Run()

	ackDispatcher, err := NewAckDispatcher(om, o)
	// Atm, NewAckDispatcher only returns nil err since there are no
	// error triggering conditions in NewAckDispatcher
	if err != nil {
		return nil, err
	}

	return []device.Listener{eventDispatcher.OnDeviceEvent, ackDispatcher.OnDeviceEvent}, nil
}