daemon/logger: remove uses of pkg/urlutil, fix fluentd validation and parsing
### daemon/logger/gelf: remove uses of pkg/urlutil.IsTransportURL()

pkg/urlutil (despite its poorly chosen name) is not really intended as a generic utility to handle URLs, and should only be used by the builder to handle (remote) build contexts.

This patch:

- removes a redundant use of urlutil.IsTransportURL(); code further below already checked if the given scheme (protocol) was supported.
- renames some variables that collided with imported packages.

### daemon/logger/splunk: remove uses of pkg/urlutil.IsURL()

pkg/urlutil (despite its poorly chosen name) is not really intended as a generic utility to handle URLs, and should only be used by the builder to handle (remote) build contexts.

This patch removes the use of urlutil.IsURL(), in favor of just checking if the provided scheme (protocol) is supported.

### daemon/logger/syslog: remove uses of pkg/urlutil.IsTransportURL()

pkg/urlutil (despite its poorly chosen name) is not really intended as a generic utility to handle URLs, and should only be used by the builder to handle (remote) build contexts.

This patch:

- removes a redundant use of urlutil.IsTransportURL(); instead adding some code to check if the given scheme (protocol) is supported.
- define a `defaultPort` const for the default port.
- use `net.JoinHostPort()` instead of string concatenating, to account for possible issues with IPv6 addresses.
- renames a variable that collided with an imported package.


### daemon/logger/fluentd: add coverage for ValidateLogOpt(), parseAddress()

This exposed a bug where host is ignored on some valid cases (to be fixed).

### daemon/logger/fluentd: rename var that collided with import

### daemon/logger/fluentd: make error-handling less DRY

### daemon/logger/fluentd: validate path element

fix some missing validation: the driver was silently ignoring path elements in some cases, and expecting a host (not an URL), and for unix sockets did not validate if a path was specified.

For the latter case, we should have a fix in the upstream driver, as it uses an empty path as default path for the socket (`defaultSocketPath`), and performs no validation.

### daemon/logger/fluentd: fix missing host, remove urlutil.IsTransportURL()

pkg/urlutil (despite its poorly chosen name) is not really intended as a generic utility to handle URLs, and should only be used by the builder to handle (remote) build contexts.

This patch:

- fix some cases where the host was ignored for valid addresses.
- removes a redundant use of urlutil.IsTransportURL(); instead adding code to check if the given scheme (protocol) is supported.
- improve port validation for out of range ports.
- fix some missing validation: the driver was silently ignoring path elements, but expected a host (not an URL)

### daemon/logger/fluentd: remove udp, tcp+tls, unixgram, add tls scheme

unix and unixgram were added in cb176c848e0731f77fa48b4e1a90ae74d1f2deae (https://github.com/moby/moby/pull/26088), but at the time, the driver only supported "tcp" and "unix": see [vendor/src/github.com/fluent/fluent-logger-golang/fluent/fluent.go#L243-L2610](https://github.com/moby/moby/blob/cb176c848e0731f77fa48b4e1a90ae74d1f2deae/vendor/src/github.com/fluent/fluent-logger-golang/fluent/fluent.go#L243-L2610)

support for tls was added in github.com/fluent/fluent-logger-golang v1.8.0, which was vendored in e24d61b7efac787ff3d5176d994608937a057522 (https://github.com/moby/moby/pull/42979).

the list of currently supported schemes by the driver is: tcp, tls and unix: see [vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go#L435-L463](https://github.com/moby/moby/blob/5179299b98a37ac2ba6eb8aefe8bf211f33f7dcc/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go#L435-L463)

daemon/logger/fluentd/fluentd.go
// Package fluentd provides the log driver for forwarding server logs
// to fluentd endpoints.
package fluentd // import "github.com/docker/docker/daemon/logger/fluentd"

import (
	"math"
	"net"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/docker/docker/daemon/logger"
	"github.com/docker/docker/daemon/logger/loggerutils"
	"github.com/docker/docker/errdefs"
	"github.com/docker/docker/pkg/urlutil"
	units "github.com/docker/go-units"
	"github.com/fluent/fluent-logger-golang/fluent"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

type fluentd struct {
	tag           string
	containerID   string
	containerName string
	writer        *fluent.Fluent
	extra         map[string]string
}

type location struct {
	protocol string
	host     string
	port     int
	path     string
}

const (
	name = "fluentd"

	defaultBufferLimit = 1024 * 1024
	defaultHost        = "127.0.0.1"
	defaultPort        = 24224
	defaultProtocol    = "tcp"

	// logger tries to reconnect 2**32 - 1 times
	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
	defaultMaxRetries = math.MaxInt32
	defaultRetryWait  = 1000

	minReconnectInterval = 100 * time.Millisecond
	maxReconnectInterval = 10 * time.Second

	addressKey                = "fluentd-address"
	asyncKey                  = "fluentd-async"
	asyncConnectKey           = "fluentd-async-connect" // deprecated option (use fluent-async instead)
	asyncReconnectIntervalKey = "fluentd-async-reconnect-interval"
	bufferLimitKey            = "fluentd-buffer-limit"
	maxRetriesKey             = "fluentd-max-retries"
	requestAckKey             = "fluentd-request-ack"
	retryWaitKey              = "fluentd-retry-wait"
	subSecondPrecisionKey     = "fluentd-sub-second-precision"
)

func init() {
	if err := logger.RegisterLogDriver(name, New); err != nil {
		logrus.Fatal(err)
	}
	if err := logger.RegisterLogOptValidator(name, ValidateLogOpt); err != nil {
		logrus.Fatal(err)
	}
}

// New creates a fluentd logger using the configuration passed in on
// the context. The supported context configuration variable is
// fluentd-address.
func New(info logger.Info) (logger.Logger, error) {
	fluentConfig, err := parseConfig(info.Config)
	if err != nil {
		return nil, errdefs.InvalidParameter(err)
	}

	tag, err := loggerutils.ParseLogTag(info, loggerutils.DefaultTemplate)
	if err != nil {
		return nil, errdefs.InvalidParameter(err)
	}

	extra, err := info.ExtraAttributes(nil)
	if err != nil {
		return nil, errdefs.InvalidParameter(err)
	}

	logrus.WithField("container", info.ContainerID).WithField("config", fluentConfig).
		Debug("logging driver fluentd configured")

	log, err := fluent.New(fluentConfig)
	if err != nil {
		return nil, err
	}
	return &fluentd{
		tag:           tag,
		containerID:   info.ContainerID,
		containerName: info.ContainerName,
		writer:        log,
		extra:         extra,
	}, nil
}

func (f *fluentd) Log(msg *logger.Message) error {
	data := map[string]string{
		"container_id":   f.containerID,
		"container_name": f.containerName,
		"source":         msg.Source,
		"log":            string(msg.Line),
	}
	for k, v := range f.extra {
		data[k] = v
	}
	if msg.PLogMetaData != nil {
		data["partial_message"] = "true"
		data["partial_id"] = msg.PLogMetaData.ID
		data["partial_ordinal"] = strconv.Itoa(msg.PLogMetaData.Ordinal)
		data["partial_last"] = strconv.FormatBool(msg.PLogMetaData.Last)
	}

	ts := msg.Timestamp
	logger.PutMessage(msg)
	// fluent-logger-golang buffers logs from failures and disconnections,
	// and these are transferred again automatically.
	return f.writer.PostWithTime(f.tag, ts, data)
}

func (f *fluentd) Close() error {
	return f.writer.Close()
}

func (f *fluentd) Name() string {
	return name
}

// ValidateLogOpt looks for fluentd specific log option fluentd-address.
func ValidateLogOpt(cfg map[string]string) error {
	for key := range cfg {
		switch key {
		case "env":
		case "env-regex":
		case "labels":
		case "labels-regex":
		case "tag":

		case addressKey:
		case asyncKey:
		case asyncConnectKey:
		case asyncReconnectIntervalKey:
		case bufferLimitKey:
		case maxRetriesKey:
		case requestAckKey:
		case retryWaitKey:
		case subSecondPrecisionKey:
			// Accepted
		default:
			return errors.Errorf("unknown log opt '%s' for fluentd log driver", key)
		}
	}

	_, err := parseConfig(cfg)
	return err
}

func parseConfig(cfg map[string]string) (fluent.Config, error) {
	var config fluent.Config

	loc, err := parseAddress(cfg[addressKey])
	if err != nil {
		return config, err
	}

	bufferLimit := defaultBufferLimit
	if cfg[bufferLimitKey] != "" {
		bl64, err := units.RAMInBytes(cfg[bufferLimitKey])
		if err != nil {
			return config, err
		}
		bufferLimit = int(bl64)
	}

	retryWait := defaultRetryWait
	if cfg[retryWaitKey] != "" {
		rwd, err := time.ParseDuration(cfg[retryWaitKey])
		if err != nil {
			return config, err
		}
		retryWait = int(rwd.Seconds() * 1000)
	}

	maxRetries := defaultMaxRetries
	if cfg[maxRetriesKey] != "" {
		mr64, err := strconv.ParseUint(cfg[maxRetriesKey], 10, strconv.IntSize)
		if err != nil {
			return config, err
		}
		maxRetries = int(mr64)
	}

	if cfg[asyncKey] != "" && cfg[asyncConnectKey] != "" {
		return config, errors.Errorf("conflicting options: cannot specify both '%s' and '%s", asyncKey, asyncConnectKey)
	}

	async := false
	if cfg[asyncKey] != "" {
		if async, err = strconv.ParseBool(cfg[asyncKey]); err != nil {
			return config, err
		}
	}

	// TODO fluentd-async-connect is deprecated in driver v1.4.0. Remove after two stable releases
	asyncConnect := false
	if cfg[asyncConnectKey] != "" {
		if asyncConnect, err = strconv.ParseBool(cfg[asyncConnectKey]); err != nil {
			return config, err
		}
	}

	asyncReconnectInterval := 0
	if cfg[asyncReconnectIntervalKey] != "" {
		interval, err := time.ParseDuration(cfg[asyncReconnectIntervalKey])
		if err != nil {
			return config, errors.Wrapf(err, "invalid value for %s", asyncReconnectIntervalKey)
		}
		if interval != 0 && (interval < minReconnectInterval || interval > maxReconnectInterval) {
			return config, errors.Errorf("invalid value for %s: value (%q) must be between %s and %s",
				asyncReconnectIntervalKey, interval, minReconnectInterval, maxReconnectInterval)
		}
		asyncReconnectInterval = int(interval.Milliseconds())
	}

	subSecondPrecision := false
	if cfg[subSecondPrecisionKey] != "" {
		if subSecondPrecision, err = strconv.ParseBool(cfg[subSecondPrecisionKey]); err != nil {
			return config, err
		}
	}

	requestAck := false
	if cfg[requestAckKey] != "" {
		if requestAck, err = strconv.ParseBool(cfg[requestAckKey]); err != nil {
			return config, err
		}
	}

	config = fluent.Config{
		FluentPort:             loc.port,
		FluentHost:             loc.host,
		FluentNetwork:          loc.protocol,
		FluentSocketPath:       loc.path,
		BufferLimit:            bufferLimit,
		RetryWait:              retryWait,
		MaxRetry:               maxRetries,
		Async:                  async,
		AsyncConnect:           asyncConnect,
		AsyncReconnectInterval: asyncReconnectInterval,
		SubSecondPrecision:     subSecondPrecision,
		RequestAck:             requestAck,
		ForceStopAsyncSend:     async || asyncConnect,
	}

	return config, nil
}

func parseAddress(address string) (*location, error) {
	if address == "" {
		return &location{
			protocol: defaultProtocol,
			host:     defaultHost,
			port:     defaultPort,
			path:     "",
		}, nil
	}

	protocol := defaultProtocol
	givenAddress := address
	if urlutil.IsTransportURL(address) {
		url, err := url.Parse(address)
		if err != nil {
			return nil, errors.Wrapf(err, "invalid fluentd-address %s", givenAddress)
		}
		// unix and unixgram socket
		if url.Scheme == "unix" || url.Scheme == "unixgram" {
			return &location{
				protocol: url.Scheme,
				host:     "",
				port:     0,
				path:     url.Path,
			}, nil
		}
		// tcp|udp
		protocol = url.Scheme
		address = url.Host
	}

	host, port, err := net.SplitHostPort(address)
	if err != nil {
		if !strings.Contains(err.Error(), "missing port in address") {
			return nil, errors.Wrapf(err, "invalid fluentd-address %s", givenAddress)
		}
		return &location{
			protocol: protocol,
			host:     host,
			port:     defaultPort,
			path:     "",
		}, nil
	}

	portnum, err := strconv.Atoi(port)
	if err != nil {
		return nil, errors.Wrapf(err, "invalid fluentd-address %s", givenAddress)
	}
	return &location{
		protocol: protocol,
		host:     host,
		port:     portnum,
		path:     "",
	}, nil
}

// Package fluentd provides the log driver for forwarding server logs
// to fluentd endpoints.
package fluentd // import "github.com/docker/docker/daemon/logger/fluentd"

import (
	"math"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/docker/docker/daemon/logger"
	"github.com/docker/docker/daemon/logger/loggerutils"
	"github.com/docker/docker/errdefs"
	units "github.com/docker/go-units"
	"github.com/fluent/fluent-logger-golang/fluent"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

type fluentd struct {
	tag           string
	containerID   string
	containerName string
	writer        *fluent.Fluent
	extra         map[string]string
}

type location struct {
	protocol string
	host     string
	port     int
	path     string
}

const (
	name = "fluentd"

	defaultBufferLimit = 1024 * 1024
	defaultHost        = "127.0.0.1"
	defaultPort        = 24224
	defaultProtocol    = "tcp"

	// logger tries to reconnect 2**32 - 1 times
	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
	defaultMaxRetries = math.MaxInt32
	defaultRetryWait  = 1000

	minReconnectInterval = 100 * time.Millisecond
	maxReconnectInterval = 10 * time.Second

	addressKey                = "fluentd-address"
	asyncKey                  = "fluentd-async"
	asyncConnectKey           = "fluentd-async-connect" // deprecated option (use fluent-async instead)
	asyncReconnectIntervalKey = "fluentd-async-reconnect-interval"
	bufferLimitKey            = "fluentd-buffer-limit"
	maxRetriesKey             = "fluentd-max-retries"
	requestAckKey             = "fluentd-request-ack"
	retryWaitKey              = "fluentd-retry-wait"
	subSecondPrecisionKey     = "fluentd-sub-second-precision"
)

func init() {
	if err := logger.RegisterLogDriver(name, New); err != nil {
		logrus.Fatal(err)
	}
	if err := logger.RegisterLogOptValidator(name, ValidateLogOpt); err != nil {
		logrus.Fatal(err)
	}
}

// New creates a fluentd logger using the configuration passed in on
// the context. The supported context configuration variable is
// fluentd-address.
func New(info logger.Info) (logger.Logger, error) {
	fluentConfig, err := parseConfig(info.Config)
	if err != nil {
		return nil, errdefs.InvalidParameter(err)
	}

	tag, err := loggerutils.ParseLogTag(info, loggerutils.DefaultTemplate)
	if err != nil {
		return nil, errdefs.InvalidParameter(err)
	}

	extra, err := info.ExtraAttributes(nil)
	if err != nil {
		return nil, errdefs.InvalidParameter(err)
	}

	logrus.WithField("container", info.ContainerID).WithField("config", fluentConfig).
		Debug("logging driver fluentd configured")

	log, err := fluent.New(fluentConfig)
	if err != nil {
		return nil, err
	}
	return &fluentd{
		tag:           tag,
		containerID:   info.ContainerID,
		containerName: info.ContainerName,
		writer:        log,
		extra:         extra,
	}, nil
}

func (f *fluentd) Log(msg *logger.Message) error {
	data := map[string]string{
		"container_id":   f.containerID,
		"container_name": f.containerName,
		"source":         msg.Source,
		"log":            string(msg.Line),
	}
	for k, v := range f.extra {
		data[k] = v
	}
	if msg.PLogMetaData != nil {
		data["partial_message"] = "true"
		data["partial_id"] = msg.PLogMetaData.ID
		data["partial_ordinal"] = strconv.Itoa(msg.PLogMetaData.Ordinal)
		data["partial_last"] = strconv.FormatBool(msg.PLogMetaData.Last)
	}

	ts := msg.Timestamp
	logger.PutMessage(msg)
	// fluent-logger-golang buffers logs from failures and disconnections,
	// and these are transferred again automatically.
	return f.writer.PostWithTime(f.tag, ts, data)
}

func (f *fluentd) Close() error {
	return f.writer.Close()
}

func (f *fluentd) Name() string {
	return name
}

// ValidateLogOpt looks for fluentd specific log option fluentd-address.
func ValidateLogOpt(cfg map[string]string) error {
	for key := range cfg {
		switch key {
		case "env":
		case "env-regex":
		case "labels":
		case "labels-regex":
		case "tag":

		case addressKey:
		case asyncKey:
		case asyncConnectKey:
		case asyncReconnectIntervalKey:
		case bufferLimitKey:
		case maxRetriesKey:
		case requestAckKey:
		case retryWaitKey:
		case subSecondPrecisionKey:
			// Accepted
		default:
			return errors.Errorf("unknown log opt '%s' for fluentd log driver", key)
		}
	}

	_, err := parseConfig(cfg)
	return err
}

func parseConfig(cfg map[string]string) (fluent.Config, error) {
	var config fluent.Config

	loc, err := parseAddress(cfg[addressKey])
	if err != nil {
		return config, errors.Wrapf(err, "invalid fluentd-address (%s)", cfg[addressKey])
	}

	bufferLimit := defaultBufferLimit
	if cfg[bufferLimitKey] != "" {
		bl64, err := units.RAMInBytes(cfg[bufferLimitKey])
		if err != nil {
			return config, err
		}
		bufferLimit = int(bl64)
	}

	retryWait := defaultRetryWait
	if cfg[retryWaitKey] != "" {
		rwd, err := time.ParseDuration(cfg[retryWaitKey])
		if err != nil {
			return config, err
		}
		retryWait = int(rwd.Seconds() * 1000)
	}

	maxRetries := defaultMaxRetries
	if cfg[maxRetriesKey] != "" {
		mr64, err := strconv.ParseUint(cfg[maxRetriesKey], 10, strconv.IntSize)
		if err != nil {
			return config, err
		}
		maxRetries = int(mr64)
	}

	if cfg[asyncKey] != "" && cfg[asyncConnectKey] != "" {
		return config, errors.Errorf("conflicting options: cannot specify both '%s' and '%s", asyncKey, asyncConnectKey)
	}

	async := false
	if cfg[asyncKey] != "" {
		if async, err = strconv.ParseBool(cfg[asyncKey]); err != nil {
			return config, err
		}
	}

	// TODO fluentd-async-connect is deprecated in driver v1.4.0. Remove after two stable releases
	asyncConnect := false
	if cfg[asyncConnectKey] != "" {
		if asyncConnect, err = strconv.ParseBool(cfg[asyncConnectKey]); err != nil {
			return config, err
		}
	}

	asyncReconnectInterval := 0
	if cfg[asyncReconnectIntervalKey] != "" {
		interval, err := time.ParseDuration(cfg[asyncReconnectIntervalKey])
		if err != nil {
			return config, errors.Wrapf(err, "invalid value for %s", asyncReconnectIntervalKey)
		}
		if interval != 0 && (interval < minReconnectInterval || interval > maxReconnectInterval) {
			return config, errors.Errorf("invalid value for %s: value (%q) must be between %s and %s",
				asyncReconnectIntervalKey, interval, minReconnectInterval, maxReconnectInterval)
		}
		asyncReconnectInterval = int(interval.Milliseconds())
	}

	subSecondPrecision := false
	if cfg[subSecondPrecisionKey] != "" {
		if subSecondPrecision, err = strconv.ParseBool(cfg[subSecondPrecisionKey]); err != nil {
			return config, err
		}
	}

	requestAck := false
	if cfg[requestAckKey] != "" {
		if requestAck, err = strconv.ParseBool(cfg[requestAckKey]); err != nil {
			return config, err
		}
	}

	config = fluent.Config{
		FluentPort:             loc.port,
		FluentHost:             loc.host,
		FluentNetwork:          loc.protocol,
		FluentSocketPath:       loc.path,
		BufferLimit:            bufferLimit,
		RetryWait:              retryWait,
		MaxRetry:               maxRetries,
		Async:                  async,
		AsyncConnect:           asyncConnect,
		AsyncReconnectInterval: asyncReconnectInterval,
		SubSecondPrecision:     subSecondPrecision,
		RequestAck:             requestAck,
		ForceStopAsyncSend:     async || asyncConnect,
	}

	return config, nil
}

func parseAddress(address string) (*location, error) {
	if address == "" {
		return &location{
			protocol: defaultProtocol,
			host:     defaultHost,
			port:     defaultPort,
			path:     "",
		}, nil
	}

	if !strings.Contains(address, "://") {
		address = defaultProtocol + "://" + address
	}

	addr, err := url.Parse(address)
	if err != nil {
		return nil, err
	}

	switch addr.Scheme {
	case "unix":
		if strings.TrimLeft(addr.Path, "/") == "" {
			return nil, errors.New("path is empty")
		}
		return &location{protocol: addr.Scheme, path: addr.Path}, nil
	case "tcp", "tls":
		// continue processing below
	default:
		return nil, errors.Errorf("unsupported scheme: '%s'", addr.Scheme)
	}

	if addr.Path != "" {
		return nil, errors.New("should not contain a path element")
	}

	host := defaultHost
	port := defaultPort

	if h := addr.Hostname(); h != "" {
		host = h
	}
	if p := addr.Port(); p != "" {
		// Port numbers are 16 bit: https://www.ietf.org/rfc/rfc793.html#section-3.1
		portNum, err := strconv.ParseUint(p, 10, 16)
		if err != nil {
			return nil, errors.Wrap(err, "invalid port")
		}
		port = int(portNum)
	}
	return &location{
		protocol: addr.Scheme,
		host:     host,
		port:     port,
		path:     "",
	}, nil
}

based on the vendor code, it looks like it never worked (I was initially considering the same) https://github.com/moby/moby/blob/5179299b98a37ac2ba6eb8aefe8bf211f33f7dcc/vendor/github.com/fluent/fluent-logger-golang/fluent/fluent.go#L435-L463