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/gelf/gelf.go
// Package gelf provides the log driver for forwarding server logs to
// endpoints that support the Graylog Extended Log Format.
package gelf // import "github.com/docker/docker/daemon/logger/gelf"

import (
	"compress/flate"
	"encoding/json"
	"fmt"
	"net"
	"net/url"
	"strconv"
	"time"

	"github.com/Graylog2/go-gelf/gelf"
	"github.com/docker/docker/daemon/logger"
	"github.com/docker/docker/daemon/logger/loggerutils"
	"github.com/docker/docker/pkg/urlutil"
	"github.com/sirupsen/logrus"
)

const name = "gelf"

type gelfLogger struct {
	writer   gelf.Writer
	info     logger.Info
	hostname string
	rawExtra json.RawMessage
}

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 gelf logger using the configuration passed in on the
// context. The supported context configuration variable is gelf-address.
func New(info logger.Info) (logger.Logger, error) {
	// parse gelf address
	address, err := parseAddress(info.Config["gelf-address"])
	if err != nil {
		return nil, err
	}

	// collect extra data for GELF message
	hostname, err := info.Hostname()
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot access hostname to set source field")
	}

	// parse log tag
	tag, err := loggerutils.ParseLogTag(info, loggerutils.DefaultTemplate)
	if err != nil {
		return nil, err
	}

	extra := map[string]interface{}{
		"_container_id":   info.ContainerID,
		"_container_name": info.Name(),
		"_image_id":       info.ContainerImageID,
		"_image_name":     info.ContainerImageName,
		"_command":        info.Command(),
		"_tag":            tag,
		"_created":        info.ContainerCreated,
	}

	extraAttrs, err := info.ExtraAttributes(func(key string) string {
		if key[0] == '_' {
			return key
		}
		return "_" + key
	})

	if err != nil {
		return nil, err
	}

	for k, v := range extraAttrs {
		extra[k] = v
	}

	rawExtra, err := json.Marshal(extra)
	if err != nil {
		return nil, err
	}

	var gelfWriter gelf.Writer
	if address.Scheme == "udp" {
		gelfWriter, err = newGELFUDPWriter(address.Host, info)
		if err != nil {
			return nil, err
		}
	} else if address.Scheme == "tcp" {
		gelfWriter, err = newGELFTCPWriter(address.Host, info)
		if err != nil {
			return nil, err
		}
	}

	return &gelfLogger{
		writer:   gelfWriter,
		info:     info,
		hostname: hostname,
		rawExtra: rawExtra,
	}, nil
}

// create new TCP gelfWriter
func newGELFTCPWriter(address string, info logger.Info) (gelf.Writer, error) {
	gelfWriter, err := gelf.NewTCPWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	if v, ok := info.Config["gelf-tcp-max-reconnect"]; ok {
		i, err := strconv.Atoi(v)
		if err != nil || i < 0 {
			return nil, fmt.Errorf("gelf-tcp-max-reconnect must be a positive integer")
		}
		gelfWriter.MaxReconnect = i
	}

	if v, ok := info.Config["gelf-tcp-reconnect-delay"]; ok {
		i, err := strconv.Atoi(v)
		if err != nil || i < 0 {
			return nil, fmt.Errorf("gelf-tcp-reconnect-delay must be a positive integer")
		}
		gelfWriter.ReconnectDelay = time.Duration(i)
	}

	return gelfWriter, nil
}

// create new UDP gelfWriter
func newGELFUDPWriter(address string, info logger.Info) (gelf.Writer, error) {
	gelfWriter, err := gelf.NewUDPWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	if v, ok := info.Config["gelf-compression-type"]; ok {
		switch v {
		case "gzip":
			gelfWriter.CompressionType = gelf.CompressGzip
		case "zlib":
			gelfWriter.CompressionType = gelf.CompressZlib
		case "none":
			gelfWriter.CompressionType = gelf.CompressNone
		default:
			return nil, fmt.Errorf("gelf: invalid compression type %q", v)
		}
	}

	if v, ok := info.Config["gelf-compression-level"]; ok {
		val, err := strconv.Atoi(v)
		if err != nil {
			return nil, fmt.Errorf("gelf: invalid compression level %s, err %v", v, err)
		}
		gelfWriter.CompressionLevel = val
	}

	return gelfWriter, nil
}

func (s *gelfLogger) Log(msg *logger.Message) error {
	if len(msg.Line) == 0 {
		return nil
	}

	level := gelf.LOG_INFO
	if msg.Source == "stderr" {
		level = gelf.LOG_ERR
	}

	m := gelf.Message{
		Version:  "1.1",
		Host:     s.hostname,
		Short:    string(msg.Line),
		TimeUnix: float64(msg.Timestamp.UnixNano()/int64(time.Millisecond)) / 1000.0,
		Level:    int32(level),
		RawExtra: s.rawExtra,
	}
	logger.PutMessage(msg)

	if err := s.writer.WriteMessage(&m); err != nil {
		return fmt.Errorf("gelf: cannot send GELF message: %v", err)
	}
	return nil
}

func (s *gelfLogger) Close() error {
	return s.writer.Close()
}

func (s *gelfLogger) Name() string {
	return name
}

// ValidateLogOpt looks for gelf specific log option gelf-address.
func ValidateLogOpt(cfg map[string]string) error {
	address, err := parseAddress(cfg["gelf-address"])
	if err != nil {
		return err
	}

	for key, val := range cfg {
		switch key {
		case "gelf-address":
		case "tag":
		case "labels":
		case "labels-regex":
		case "env":
		case "env-regex":
		case "gelf-compression-level":
			if address.Scheme != "udp" {
				return fmt.Errorf("compression is only supported on UDP")
			}
			i, err := strconv.Atoi(val)
			if err != nil || i < flate.DefaultCompression || i > flate.BestCompression {
				return fmt.Errorf("unknown value %q for log opt %q for gelf log driver", val, key)
			}
		case "gelf-compression-type":
			if address.Scheme != "udp" {
				return fmt.Errorf("compression is only supported on UDP")
			}
			switch val {
			case "gzip", "zlib", "none":
			default:
				return fmt.Errorf("unknown value %q for log opt %q for gelf log driver", val, key)
			}
		case "gelf-tcp-max-reconnect", "gelf-tcp-reconnect-delay":
			if address.Scheme != "tcp" {
				return fmt.Errorf("%q is only valid for TCP", key)
			}
			i, err := strconv.Atoi(val)
			if err != nil || i < 0 {
				return fmt.Errorf("%q must be a positive integer", key)
			}
		default:
			return fmt.Errorf("unknown log opt %q for gelf log driver", key)
		}
	}

	return nil
}

func parseAddress(address string) (*url.URL, error) {
	if address == "" {
		return nil, fmt.Errorf("gelf-address is a required parameter")
	}
	if !urlutil.IsTransportURL(address) {
		return nil, fmt.Errorf("gelf-address should be in form proto://address, got %v", address)
	}
	url, err := url.Parse(address)
	if err != nil {
		return nil, err
	}

	// we support only udp
	if url.Scheme != "udp" && url.Scheme != "tcp" {
		return nil, fmt.Errorf("gelf: endpoint needs to be TCP or UDP")
	}

	// get host and port
	if _, _, err = net.SplitHostPort(url.Host); err != nil {
		return nil, fmt.Errorf("gelf: please provide gelf-address as proto://host:port")
	}

	return url, nil
}

// Package gelf provides the log driver for forwarding server logs to
// endpoints that support the Graylog Extended Log Format.
package gelf // import "github.com/docker/docker/daemon/logger/gelf"

import (
	"compress/flate"
	"encoding/json"
	"fmt"
	"net"
	"net/url"
	"strconv"
	"time"

	"github.com/Graylog2/go-gelf/gelf"
	"github.com/docker/docker/daemon/logger"
	"github.com/docker/docker/daemon/logger/loggerutils"
	"github.com/sirupsen/logrus"
)

const name = "gelf"

type gelfLogger struct {
	writer   gelf.Writer
	info     logger.Info
	hostname string
	rawExtra json.RawMessage
}

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 gelf logger using the configuration passed in on the
// context. The supported context configuration variable is gelf-address.
func New(info logger.Info) (logger.Logger, error) {
	// parse gelf address
	address, err := parseAddress(info.Config["gelf-address"])
	if err != nil {
		return nil, err
	}

	// collect extra data for GELF message
	hostname, err := info.Hostname()
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot access hostname to set source field")
	}

	// parse log tag
	tag, err := loggerutils.ParseLogTag(info, loggerutils.DefaultTemplate)
	if err != nil {
		return nil, err
	}

	extra := map[string]interface{}{
		"_container_id":   info.ContainerID,
		"_container_name": info.Name(),
		"_image_id":       info.ContainerImageID,
		"_image_name":     info.ContainerImageName,
		"_command":        info.Command(),
		"_tag":            tag,
		"_created":        info.ContainerCreated,
	}

	extraAttrs, err := info.ExtraAttributes(func(key string) string {
		if key[0] == '_' {
			return key
		}
		return "_" + key
	})

	if err != nil {
		return nil, err
	}

	for k, v := range extraAttrs {
		extra[k] = v
	}

	rawExtra, err := json.Marshal(extra)
	if err != nil {
		return nil, err
	}

	var gelfWriter gelf.Writer
	if address.Scheme == "udp" {
		gelfWriter, err = newGELFUDPWriter(address.Host, info)
		if err != nil {
			return nil, err
		}
	} else if address.Scheme == "tcp" {
		gelfWriter, err = newGELFTCPWriter(address.Host, info)
		if err != nil {
			return nil, err
		}
	}

	return &gelfLogger{
		writer:   gelfWriter,
		info:     info,
		hostname: hostname,
		rawExtra: rawExtra,
	}, nil
}

// create new TCP gelfWriter
func newGELFTCPWriter(address string, info logger.Info) (gelf.Writer, error) {
	gelfWriter, err := gelf.NewTCPWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	if v, ok := info.Config["gelf-tcp-max-reconnect"]; ok {
		i, err := strconv.Atoi(v)
		if err != nil || i < 0 {
			return nil, fmt.Errorf("gelf-tcp-max-reconnect must be a positive integer")
		}
		gelfWriter.MaxReconnect = i
	}

	if v, ok := info.Config["gelf-tcp-reconnect-delay"]; ok {
		i, err := strconv.Atoi(v)
		if err != nil || i < 0 {
			return nil, fmt.Errorf("gelf-tcp-reconnect-delay must be a positive integer")
		}
		gelfWriter.ReconnectDelay = time.Duration(i)
	}

	return gelfWriter, nil
}

// create new UDP gelfWriter
func newGELFUDPWriter(address string, info logger.Info) (gelf.Writer, error) {
	gelfWriter, err := gelf.NewUDPWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	if v, ok := info.Config["gelf-compression-type"]; ok {
		switch v {
		case "gzip":
			gelfWriter.CompressionType = gelf.CompressGzip
		case "zlib":
			gelfWriter.CompressionType = gelf.CompressZlib
		case "none":
			gelfWriter.CompressionType = gelf.CompressNone
		default:
			return nil, fmt.Errorf("gelf: invalid compression type %q", v)
		}
	}

	if v, ok := info.Config["gelf-compression-level"]; ok {
		val, err := strconv.Atoi(v)
		if err != nil {
			return nil, fmt.Errorf("gelf: invalid compression level %s, err %v", v, err)
		}
		gelfWriter.CompressionLevel = val
	}

	return gelfWriter, nil
}

func (s *gelfLogger) Log(msg *logger.Message) error {
	if len(msg.Line) == 0 {
		return nil
	}

	level := gelf.LOG_INFO
	if msg.Source == "stderr" {
		level = gelf.LOG_ERR
	}

	m := gelf.Message{
		Version:  "1.1",
		Host:     s.hostname,
		Short:    string(msg.Line),
		TimeUnix: float64(msg.Timestamp.UnixNano()/int64(time.Millisecond)) / 1000.0,
		Level:    int32(level),
		RawExtra: s.rawExtra,
	}
	logger.PutMessage(msg)

	if err := s.writer.WriteMessage(&m); err != nil {
		return fmt.Errorf("gelf: cannot send GELF message: %v", err)
	}
	return nil
}

func (s *gelfLogger) Close() error {
	return s.writer.Close()
}

func (s *gelfLogger) Name() string {
	return name
}

// ValidateLogOpt looks for gelf specific log option gelf-address.
func ValidateLogOpt(cfg map[string]string) error {
	address, err := parseAddress(cfg["gelf-address"])
	if err != nil {
		return err
	}

	for key, val := range cfg {
		switch key {
		case "gelf-address":
		case "tag":
		case "labels":
		case "labels-regex":
		case "env":
		case "env-regex":
		case "gelf-compression-level":
			if address.Scheme != "udp" {
				return fmt.Errorf("compression is only supported on UDP")
			}
			i, err := strconv.Atoi(val)
			if err != nil || i < flate.DefaultCompression || i > flate.BestCompression {
				return fmt.Errorf("unknown value %q for log opt %q for gelf log driver", val, key)
			}
		case "gelf-compression-type":
			if address.Scheme != "udp" {
				return fmt.Errorf("compression is only supported on UDP")
			}
			switch val {
			case "gzip", "zlib", "none":
			default:
				return fmt.Errorf("unknown value %q for log opt %q for gelf log driver", val, key)
			}
		case "gelf-tcp-max-reconnect", "gelf-tcp-reconnect-delay":
			if address.Scheme != "tcp" {
				return fmt.Errorf("%q is only valid for TCP", key)
			}
			i, err := strconv.Atoi(val)
			if err != nil || i < 0 {
				return fmt.Errorf("%q must be a positive integer", key)
			}
		default:
			return fmt.Errorf("unknown log opt %q for gelf log driver", key)
		}
	}

	return nil
}

func parseAddress(address string) (*url.URL, error) {
	if address == "" {
		return nil, fmt.Errorf("gelf-address is a required parameter")
	}
	addr, err := url.Parse(address)
	if err != nil {
		return nil, err
	}

	if addr.Scheme != "udp" && addr.Scheme != "tcp" {
		return nil, fmt.Errorf("gelf: endpoint needs to be TCP or UDP")
	}

	if _, _, err = net.SplitHostPort(addr.Host); err != nil {
		return nil, fmt.Errorf("gelf: please provide gelf-address as proto://host:port")
	}

	return addr, nil
}

I think we should be fine; couple of lines down we check if the scheme is supported;

```go
if addr.Scheme != "udp" && addr.Scheme != "tcp" {
```

I can't come up with cases where it could invalidate a _valid_ address (at most one that's specified without scheme, but that should produce an error message, helping the user to find the problem)