// New creates a new logger that logs to Google Cloud Logging using the application
// default credentials.
//
// See https://developers.google.com/identity/protocols/application-default-credentials
func New(ctx logger.Context) (logger.Logger, error) {
	initGCP()

	var project string
	if projectID != "" {
		project = projectID
	}
	if projectID, found := ctx.Config[projectOptKey]; found {
		project = projectID
	}
	if project == "" {
		return nil, fmt.Errorf("No project was specified and couldn't read project from the meatadata server. Please specify a project")
	}

	c, err := logging.NewClient(context.Background(), project, "gcplogs-docker-driver")
	if err != nil {
		return nil, err
	}

	if err := c.Ping(); err != nil {
		return nil, fmt.Errorf("unable to connect or authenticate with Google Cloud Logging: %v", err)
	}

	l := &gcplogs{
		client: c,
		container: &containerInfo{
			Name:      ctx.ContainerName,
			ID:        ctx.ContainerID,
			ImageName: ctx.ContainerImageName,
			ImageID:   ctx.ContainerImageID,
			Created:   ctx.ContainerCreated,
			Metadata:  ctx.ExtraAttributes(nil),
		},
	}

	if ctx.Config[logCmdKey] == "true" {
		l.container.Command = ctx.Command()
	}

	if onGCE {
		l.instance = &instanceInfo{
			Zone: zone,
			Name: instanceName,
			ID:   instanceID,
		}
	}

	// The logger "overflows" at a rate of 10,000 logs per second and this
	// overflow func is called. We want to surface the error to the user
	// without overly spamming /var/log/docker.log so we log the first time
	// we overflow and every 1000th time after.
	c.Overflow = func(_ *logging.Client, _ logging.Entry) error {
		if i := atomic.AddUint64(&droppedLogs, 1); i%1000 == 1 {
			logrus.Errorf("gcplogs driver has dropped %v logs", i)
		}
		return nil
	}

	return l, nil
}

// New creates a fluentd logger using the configuration passed in on
// the context. Supported context configuration variables are
// fluentd-address & fluentd-tag.
func New(ctx logger.Context) (logger.Logger, error) {
	host, port, err := parseAddress(ctx.Config["fluentd-address"])
	if err != nil {
		return nil, err
	}

	tag, err := loggerutils.ParseLogTag(ctx, "docker.{{.ID}}")
	if err != nil {
		return nil, err
	}
	extra := ctx.ExtraAttributes(nil)
	logrus.Debugf("logging driver fluentd configured for container:%s, host:%s, port:%d, tag:%s, extra:%v.", ctx.ContainerID, host, port, tag, extra)
	// logger tries to reconnect 2**32 - 1 times
	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
	log, err := fluent.New(fluent.Config{FluentPort: port, FluentHost: host, RetryWait: 1000, MaxRetry: math.MaxInt32})
	if err != nil {
		return nil, err
	}
	return &fluentd{
		tag:           tag,
		containerID:   ctx.ContainerID,
		containerName: ctx.ContainerName,
		writer:        log,
		extra:         extra,
	}, nil
}

// New creates a journald logger using the configuration passed in on
// the context.
func New(ctx logger.Context) (logger.Logger, error) {
	if !journal.Enabled() {
		return nil, fmt.Errorf("journald is not enabled on this host")
	}
	// Strip a leading slash so that people can search for
	// CONTAINER_NAME=foo rather than CONTAINER_NAME=/foo.
	name := ctx.ContainerName
	if name[0] == '/' {
		name = name[1:]
	}

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

	vars := map[string]string{
		"CONTAINER_ID":      ctx.ContainerID[:12],
		"CONTAINER_ID_FULL": ctx.ContainerID,
		"CONTAINER_NAME":    name,
		"CONTAINER_TAG":     tag,
	}
	extraAttrs := ctx.ExtraAttributes(sanitizeKeyMod)
	for k, v := range extraAttrs {
		vars[k] = v
	}
	return &journald{vars: vars, readers: readerList{readers: make(map[*logger.LogWatcher]*logger.LogWatcher)}}, nil
}

// New creates a gelf logger using the configuration passed in on the
// context. Supported context configuration variables are
// gelf-address, & gelf-tag.
func New(ctx logger.Context) (logger.Logger, error) {
	// parse gelf address
	address, err := parseAddress(ctx.Config["gelf-address"])
	if err != nil {
		return nil, err
	}

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

	// remove trailing slash from container name
	containerName := bytes.TrimLeft([]byte(ctx.ContainerName), "/")

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

	fields := gelfFields{
		hostname:      hostname,
		containerID:   ctx.ContainerID,
		containerName: string(containerName),
		imageID:       ctx.ContainerImageID,
		imageName:     ctx.ContainerImageName,
		command:       ctx.Command(),
		tag:           tag,
		created:       ctx.ContainerCreated,
	}

	// create new gelfWriter
	gelfWriter, err := gelf.NewWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	return &gelfLogger{
		writer: gelfWriter,
		ctx:    ctx,
		fields: fields,
	}, nil
}

// New creates new JSONFileLogger which writes to filename passed in
// on given context.
func New(ctx logger.Context) (logger.Logger, error) {
	log, err := os.OpenFile(ctx.LogPath, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0600)
	if err != nil {
		return nil, err
	}
	var capval int64 = -1
	if capacity, ok := ctx.Config["max-size"]; ok {
		var err error
		capval, err = units.FromHumanSize(capacity)
		if err != nil {
			return nil, err
		}
	}
	var maxFiles = 1
	if maxFileString, ok := ctx.Config["max-file"]; ok {
		maxFiles, err = strconv.Atoi(maxFileString)
		if err != nil {
			return nil, err
		}
		if maxFiles < 1 {
			return nil, fmt.Errorf("max-file cannot be less than 1")
		}
	}

	var extra []byte
	if attrs := ctx.ExtraAttributes(nil); len(attrs) > 0 {
		var err error
		extra, err = json.Marshal(attrs)
		if err != nil {
			return nil, err
		}
	}

	return &JSONFileLogger{
		f:            log,
		buf:          bytes.NewBuffer(nil),
		ctx:          ctx,
		capacity:     capval,
		n:            maxFiles,
		readers:      make(map[*logger.LogWatcher]struct{}),
		notifyRotate: pubsub.NewPublisher(0, 1),
		extra:        extra,
	}, nil
}

func (c *Container) initLogger() {
	if c.logger.Driver != nil {
		return
	}

	if c.p.factory.logCreator == nil {
		return
	}

	ctx := logger.Context{
		Config:              c.p.factory.logCfg.Config,
		ContainerID:         c.Id(),
		ContainerName:       c.RuntimeName(),
		ContainerImageName:  c.descript.Image,
		ContainerCreated:    c.status.CreatedAt,
		ContainerEntrypoint: c.descript.Path,
		ContainerArgs:       c.descript.Args,
		ContainerImageID:    c.descript.Image,
	}

	if c.p.factory.logCfg.Type == jsonfilelog.Name {
		prefix := c.p.factory.logCfg.PathPrefix
		if c.p.factory.logCfg.PodIdInPath {
			prefix = filepath.Join(prefix, c.p.Id())
		}
		if err := os.MkdirAll(prefix, os.FileMode(0755)); err != nil {
			c.Log(ERROR, "cannot create container log dir %s: %v", prefix, err)
			return
		}

		ctx.LogPath = filepath.Join(prefix, fmt.Sprintf("%s-json.log", c.Id()))
		c.Log(DEBUG, "configure container log to %s", ctx.LogPath)
	}

	driver, err := c.p.factory.logCreator(ctx)
	if err != nil {
		return
	}
	c.logger.Driver = driver
	c.Log(DEBUG, "container logger configured")

	return
}

// New creates new JSONFileLogger which writes to filename passed in
// on given context.
func New(ctx logger.Context) (logger.Logger, error) {
	var capval int64 = -1
	if capacity, ok := ctx.Config["max-size"]; ok {
		var err error
		capval, err = units.FromHumanSize(capacity)
		if err != nil {
			return nil, err
		}
	}
	var maxFiles = 1
	if maxFileString, ok := ctx.Config["max-file"]; ok {
		var err error
		maxFiles, err = strconv.Atoi(maxFileString)
		if err != nil {
			return nil, err
		}
		if maxFiles < 1 {
			return nil, fmt.Errorf("max-file cannot be less than 1")
		}
	}

	writer, err := loggerutils.NewRotateFileWriter(ctx.LogPath, capval, maxFiles)
	if err != nil {
		return nil, err
	}

	var extra []byte
	if attrs := ctx.ExtraAttributes(nil); len(attrs) > 0 {
		var err error
		extra, err = json.Marshal(attrs)
		if err != nil {
			return nil, err
		}
	}

	return &JSONFileLogger{
		buf:           bytes.NewBuffer(nil),
		writer:        writer,
		readers:       make(map[*logger.LogWatcher]struct{}),
		extra:         extra,
		writeNotifier: pubsub.NewPublisher(0, 10),
	}, nil
}

// New creates a journald logger using the configuration passed in on
// the context.
func New(ctx logger.Context) (logger.Logger, error) {
	if !journal.Enabled() {
		return nil, fmt.Errorf("journald is not enabled on this host")
	}
	// Strip a leading slash so that people can search for
	// CONTAINER_NAME=foo rather than CONTAINER_NAME=/foo.
	name := ctx.ContainerName
	if name[0] == '/' {
		name = name[1:]
	}

	vars := map[string]string{
		"CONTAINER_ID":      ctx.ContainerID[:12],
		"CONTAINER_ID_FULL": ctx.ContainerID,
		"CONTAINER_NAME":    name,
	}
	extraAttrs := ctx.ExtraAttributes(strings.ToTitle)
	for k, v := range extraAttrs {
		vars[k] = v
	}
	return &journald{vars: vars, readers: readerList{readers: make(map[*logger.LogWatcher]*logger.LogWatcher)}}, nil
}

// New creates a journald logger using the configuration passed in on
// the context.
func New(ctx logger.Context) (logger.Logger, error) {
	if !journal.Enabled() {
		return nil, fmt.Errorf("journald is not enabled on this host")
	}

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

	vars := map[string]string{
		"CONTAINER_ID":      ctx.ContainerID[:12],
		"CONTAINER_ID_FULL": ctx.ContainerID,
		"CONTAINER_NAME":    ctx.Name(),
		"CONTAINER_TAG":     tag,
	}
	extraAttrs := ctx.ExtraAttributes(sanitizeKeyMod)
	for k, v := range extraAttrs {
		vars[k] = v
	}
	return &journald{vars: vars, readers: readerList{readers: make(map[*logger.LogWatcher]*logger.LogWatcher)}}, nil
}

// New creates a fluentd logger using the configuration passed in on
// the context. Supported context configuration variables are
// fluentd-address & fluentd-tag.
func New(ctx logger.Context) (logger.Logger, error) {
	host, port, err := parseAddress(ctx.Config["fluentd-address"])
	if err != nil {
		return nil, err
	}

	tag, err := loggerutils.ParseLogTag(ctx, "docker.{{.ID}}")
	if err != nil {
		return nil, err
	}
	failOnStartupError, err := loggerutils.ParseFailOnStartupErrorFlag(ctx)
	if err != nil {
		return nil, err
	}
	bufferLimit, err := parseBufferLimit(ctx.Config["buffer-limit"])
	if err != nil {
		return nil, err
	}
	extra := ctx.ExtraAttributes(nil)
	logrus.Debugf("logging driver fluentd configured for container:%s, host:%s, port:%d, tag:%s, extra:%v.", ctx.ContainerID, host, port, tag, extra)
	// logger tries to reconnect 2**32 - 1 times
	// failed (and panic) after 204 years [ 1.5 ** (2**32 - 1) - 1 seconds]
	log, err := fluent.New(fluent.Config{FluentPort: port, FluentHost: host, RetryWait: 1000, MaxRetry: math.MaxInt32, BufferLimit: bufferLimit})
	if err != nil {
		if failOnStartupError {
			return nil, err
		}
		logrus.Warnf("fluentd cannot connect to configured endpoint. Ignoring as instructed. Error: %q", err)
	}
	return &fluentd{
		tag:           tag,
		containerID:   ctx.ContainerID,
		containerName: ctx.ContainerName,
		writer:        log,
		extra:         extra,
	}, nil
}

// New creates a gelf logger using the configuration passed in on the
// context. Supported context configuration variables are
// gelf-address, & gelf-tag.
func New(ctx logger.Context) (logger.Logger, error) {
	// parse gelf address
	address, err := parseAddress(ctx.Config["gelf-address"])
	if err != nil {
		return nil, err
	}

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

	// remove trailing slash from container name
	containerName := bytes.TrimLeft([]byte(ctx.ContainerName), "/")

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

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

	extraAttrs := ctx.ExtraAttributes(func(key string) string {
		if key[0] == '_' {
			return key
		}
		return "_" + key
	})
	for k, v := range extraAttrs {
		extra[k] = v
	}

	// create new gelfWriter
	gelfWriter, err := gelf.NewWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	return &gelfLogger{
		writer:   gelfWriter,
		ctx:      ctx,
		hostname: hostname,
		extra:    extra,
	}, nil
}

// New creates a gelf logger using the configuration passed in on the
// context. The supported context configuration variable is gelf-address.
func New(ctx logger.Context) (logger.Logger, error) {
	// parse gelf address
	address, err := parseAddress(ctx.Config["gelf-address"])
	if err != nil {
		return nil, err
	}

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

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

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

	extraAttrs := ctx.ExtraAttributes(func(key string) string {
		if key[0] == '_' {
			return key
		}
		return "_" + key
	})
	for k, v := range extraAttrs {
		extra[k] = v
	}

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

	// create new gelfWriter
	gelfWriter, err := gelf.NewWriter(address)
	if err != nil {
		return nil, fmt.Errorf("gelf: cannot connect to GELF endpoint: %s %v", address, err)
	}

	if v, ok := ctx.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 := ctx.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 &gelfLogger{
		writer:   gelfWriter,
		ctx:      ctx,
		hostname: hostname,
		rawExtra: rawExtra,
	}, nil
}

func (p *Pod) getLogger(daemon *Daemon) (err error) {
	if p.spec.LogConfig.Type == "" {
		p.spec.LogConfig.Type = daemon.DefaultLog.Type
		p.spec.LogConfig.Config = daemon.DefaultLog.Config
	}

	if p.spec.LogConfig.Type == "none" {
		return nil
	}

	var (
		needLogger []int = []int{}
		creator    logger.Creator
	)

	for i, c := range p.status.Containers {
		if c.Logs.Driver == nil {
			needLogger = append(needLogger, i)
		}
	}

	if len(needLogger) == 0 && p.status.Status == types.S_POD_RUNNING {
		return nil
	}

	if err = logger.ValidateLogOpts(p.spec.LogConfig.Type, p.spec.LogConfig.Config); err != nil {
		return
	}
	creator, err = logger.GetLogDriver(p.spec.LogConfig.Type)
	if err != nil {
		return
	}
	glog.V(1).Infof("configuring log driver [%s] for %s", p.spec.LogConfig.Type, p.id)

	for i, c := range p.status.Containers {
		ctx := logger.Context{
			Config:             p.spec.LogConfig.Config,
			ContainerID:        c.Id,
			ContainerName:      c.Name,
			ContainerImageName: p.spec.Containers[i].Image,
			ContainerCreated:   time.Now(), //FIXME: should record creation time in PodStatus
		}

		if p.containers != nil && len(p.containers) > i {
			ctx.ContainerEntrypoint = p.containers[i].Workdir
			ctx.ContainerArgs = p.containers[i].Cmd
			ctx.ContainerImageID = p.containers[i].Image
		}

		if p.spec.LogConfig.Type == jsonfilelog.Name {
			ctx.LogPath = filepath.Join(p.status.ResourcePath, fmt.Sprintf("%s-json.log", c.Id))
			glog.V(1).Info("configure container log to ", ctx.LogPath)
		}

		if c.Logs.Driver, err = creator(ctx); err != nil {
			return
		}
		glog.V(1).Infof("configured logger for %s/%s (%s)", p.id, c.Id, c.Name)
	}

	return nil
}

// Test default settings
func TestDefault(t *testing.T) {
	hec := NewHTTPEventCollectorMock(t)

	go hec.Serve()

	ctx := logger.Context{
		Config: map[string]string{
			splunkURLKey:   hec.URL(),
			splunkTokenKey: hec.token,
		},
		ContainerID:        "containeriid",
		ContainerName:      "container_name",
		ContainerImageID:   "contaimageid",
		ContainerImageName: "container_image_name",
	}

	hostname, err := ctx.Hostname()
	if err != nil {
		t.Fatal(err)
	}

	loggerDriver, err := New(ctx)
	if err != nil {
		t.Fatal(err)
	}

	if loggerDriver.Name() != driverName {
		t.Fatal("Unexpected logger driver name")
	}

	if !hec.connectionVerified {
		t.Fatal("By default connection should be verified")
	}

	splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerInline)
	if !ok {
		t.Fatal("Unexpected Splunk Logging Driver type")
	}

	if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" ||
		splunkLoggerDriver.auth != "Splunk "+hec.token ||
		splunkLoggerDriver.nullMessage.Host != hostname ||
		splunkLoggerDriver.nullMessage.Source != "" ||
		splunkLoggerDriver.nullMessage.SourceType != "" ||
		splunkLoggerDriver.nullMessage.Index != "" ||
		splunkLoggerDriver.gzipCompression != false ||
		splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency ||
		splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize ||
		splunkLoggerDriver.bufferMaximum != defaultBufferMaximum ||
		cap(splunkLoggerDriver.stream) != defaultStreamChannelSize {
		t.Fatal("Found not default values setup in Splunk Logging Driver.")
	}

	message1Time := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("{\"a\":\"b\"}"), "stdout", message1Time, nil, false}); err != nil {
		t.Fatal(err)
	}
	message2Time := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("notajson"), "stdout", message2Time, nil, false}); err != nil {
		t.Fatal(err)
	}

	err = loggerDriver.Close()
	if err != nil {
		t.Fatal(err)
	}

	if len(hec.messages) != 2 {
		t.Fatal("Expected two messages")
	}

	if *hec.gzipEnabled {
		t.Fatal("Gzip should not be used")
	}

	message1 := hec.messages[0]
	if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) ||
		message1.Host != hostname ||
		message1.Source != "" ||
		message1.SourceType != "" ||
		message1.Index != "" {
		t.Fatalf("Unexpected values of message 1 %v", message1)
	}

	if event, err := message1.EventAsMap(); err != nil {
		t.Fatal(err)
	} else {
		if event["line"] != "{\"a\":\"b\"}" ||
			event["source"] != "stdout" ||
			event["tag"] != "containeriid" ||
			len(event) != 3 {
			t.Fatalf("Unexpected event in message %v", event)
		}
	}

	message2 := hec.messages[1]
	if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) ||
		message2.Host != hostname ||
		message2.Source != "" ||
//.........这里部分代码省略.........

// Verify that Splunk Logging Driver can accept tag="" which will allow to send raw messages
// in the same way we get them in stdout/stderr
func TestRawFormatWithoutTag(t *testing.T) {
	hec := NewHTTPEventCollectorMock(t)

	go hec.Serve()

	ctx := logger.Context{
		Config: map[string]string{
			splunkURLKey:    hec.URL(),
			splunkTokenKey:  hec.token,
			splunkFormatKey: splunkFormatRaw,
			tagKey:          "",
		},
		ContainerID:        "containeriid",
		ContainerName:      "/container_name",
		ContainerImageID:   "contaimageid",
		ContainerImageName: "container_image_name",
	}

	hostname, err := ctx.Hostname()
	if err != nil {
		t.Fatal(err)
	}

	loggerDriver, err := New(ctx)
	if err != nil {
		t.Fatal(err)
	}

	if !hec.connectionVerified {
		t.Fatal("By default connection should be verified")
	}

	splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerRaw)
	if !ok {
		t.Fatal("Unexpected Splunk Logging Driver type")
	}

	if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" ||
		splunkLoggerDriver.auth != "Splunk "+hec.token ||
		splunkLoggerDriver.nullMessage.Host != hostname ||
		splunkLoggerDriver.nullMessage.Source != "" ||
		splunkLoggerDriver.nullMessage.SourceType != "" ||
		splunkLoggerDriver.nullMessage.Index != "" ||
		splunkLoggerDriver.gzipCompression != false ||
		splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency ||
		splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize ||
		splunkLoggerDriver.bufferMaximum != defaultBufferMaximum ||
		cap(splunkLoggerDriver.stream) != defaultStreamChannelSize ||
		string(splunkLoggerDriver.prefix) != "" {
		t.Log(string(splunkLoggerDriver.prefix) + "a")
		t.Fatal("Values do not match configuration.")
	}

	message1Time := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("{\"a\":\"b\"}"), "stdout", message1Time, nil, false}); err != nil {
		t.Fatal(err)
	}
	message2Time := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("notjson"), "stdout", message2Time, nil, false}); err != nil {
		t.Fatal(err)
	}

	err = loggerDriver.Close()
	if err != nil {
		t.Fatal(err)
	}

	if len(hec.messages) != 2 {
		t.Fatal("Expected two messages")
	}

	message1 := hec.messages[0]
	if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) ||
		message1.Host != hostname ||
		message1.Source != "" ||
		message1.SourceType != "" ||
		message1.Index != "" {
		t.Fatalf("Unexpected values of message 1 %v", message1)
	}

	if event, err := message1.EventAsString(); err != nil {
		t.Fatal(err)
	} else {
		if event != "{\"a\":\"b\"}" {
			t.Fatalf("Unexpected event in message 1 %v", event)
		}
	}

	message2 := hec.messages[1]
	if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) ||
		message2.Host != hostname ||
		message2.Source != "" ||
		message2.SourceType != "" ||
		message2.Index != "" {
		t.Fatalf("Unexpected values of message 2 %v", message2)
	}

	if event, err := message2.EventAsString(); err != nil {
//.........这里部分代码省略.........

// Verify JSON format
func TestJsonFormat(t *testing.T) {
	hec := NewHTTPEventCollectorMock(t)

	go hec.Serve()

	ctx := logger.Context{
		Config: map[string]string{
			splunkURLKey:                  hec.URL(),
			splunkTokenKey:                hec.token,
			splunkFormatKey:               splunkFormatJSON,
			splunkGzipCompressionKey:      "true",
			splunkGzipCompressionLevelKey: "1",
		},
		ContainerID:        "containeriid",
		ContainerName:      "/container_name",
		ContainerImageID:   "contaimageid",
		ContainerImageName: "container_image_name",
	}

	hostname, err := ctx.Hostname()
	if err != nil {
		t.Fatal(err)
	}

	loggerDriver, err := New(ctx)
	if err != nil {
		t.Fatal(err)
	}

	if !hec.connectionVerified {
		t.Fatal("By default connection should be verified")
	}

	splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerJSON)
	if !ok {
		t.Fatal("Unexpected Splunk Logging Driver type")
	}

	if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" ||
		splunkLoggerDriver.auth != "Splunk "+hec.token ||
		splunkLoggerDriver.nullMessage.Host != hostname ||
		splunkLoggerDriver.nullMessage.Source != "" ||
		splunkLoggerDriver.nullMessage.SourceType != "" ||
		splunkLoggerDriver.nullMessage.Index != "" ||
		splunkLoggerDriver.gzipCompression != true ||
		splunkLoggerDriver.gzipCompressionLevel != gzip.BestSpeed ||
		splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency ||
		splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize ||
		splunkLoggerDriver.bufferMaximum != defaultBufferMaximum ||
		cap(splunkLoggerDriver.stream) != defaultStreamChannelSize {
		t.Fatal("Values do not match configuration.")
	}

	message1Time := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("{\"a\":\"b\"}"), "stdout", message1Time, nil, false}); err != nil {
		t.Fatal(err)
	}
	message2Time := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("notjson"), "stdout", message2Time, nil, false}); err != nil {
		t.Fatal(err)
	}

	err = loggerDriver.Close()
	if err != nil {
		t.Fatal(err)
	}

	if len(hec.messages) != 2 {
		t.Fatal("Expected two messages")
	}

	message1 := hec.messages[0]
	if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) ||
		message1.Host != hostname ||
		message1.Source != "" ||
		message1.SourceType != "" ||
		message1.Index != "" {
		t.Fatalf("Unexpected values of message 1 %v", message1)
	}

	if event, err := message1.EventAsMap(); err != nil {
		t.Fatal(err)
	} else {
		if event["line"].(map[string]interface{})["a"] != "b" ||
			event["source"] != "stdout" ||
			event["tag"] != "containeriid" ||
			len(event) != 3 {
			t.Fatalf("Unexpected event in message 1 %v", event)
		}
	}

	message2 := hec.messages[1]
	if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) ||
		message2.Host != hostname ||
		message2.Source != "" ||
		message2.SourceType != "" ||
		message2.Index != "" {
		t.Fatalf("Unexpected values of message 2 %v", message2)
	}
//.........这里部分代码省略.........

// New creates splunk logger driver using configuration passed in context
func New(ctx logger.Context) (logger.Logger, error) {
	hostname, err := ctx.Hostname()
	if err != nil {
		return nil, fmt.Errorf("%s: cannot access hostname to set source field", driverName)
	}

	// Parse and validate Splunk URL
	splunkURL, err := parseURL(ctx)
	if err != nil {
		return nil, err
	}

	// Splunk Token is required parameter
	splunkToken, ok := ctx.Config[splunkTokenKey]
	if !ok {
		return nil, fmt.Errorf("%s: %s is expected", driverName, splunkTokenKey)
	}

	tlsConfig := &tls.Config{}

	// Splunk is using autogenerated certificates by default,
	// allow users to trust them with skipping verification
	if insecureSkipVerifyStr, ok := ctx.Config[splunkInsecureSkipVerifyKey]; ok {
		insecureSkipVerify, err := strconv.ParseBool(insecureSkipVerifyStr)
		if err != nil {
			return nil, err
		}
		tlsConfig.InsecureSkipVerify = insecureSkipVerify
	}

	// If path to the root certificate is provided - load it
	if caPath, ok := ctx.Config[splunkCAPathKey]; ok {
		caCert, err := ioutil.ReadFile(caPath)
		if err != nil {
			return nil, err
		}
		caPool := x509.NewCertPool()
		caPool.AppendCertsFromPEM(caCert)
		tlsConfig.RootCAs = caPool
	}

	if caName, ok := ctx.Config[splunkCANameKey]; ok {
		tlsConfig.ServerName = caName
	}

	transport := &http.Transport{
		TLSClientConfig: tlsConfig,
	}
	client := &http.Client{
		Transport: transport,
	}

	source := ctx.Config[splunkSourceKey]
	sourceType := ctx.Config[splunkSourceTypeKey]
	index := ctx.Config[splunkIndexKey]

	var nullMessage = &splunkMessage{
		Host:       hostname,
		Source:     source,
		SourceType: sourceType,
		Index:      index,
	}

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

	attrs := ctx.ExtraAttributes(nil)

	logger := &splunkLogger{
		client:      client,
		transport:   transport,
		url:         splunkURL.String(),
		auth:        "Splunk " + splunkToken,
		nullMessage: nullMessage,
	}

	// By default we verify connection, but we allow use to skip that
	verifyConnection := true
	if verifyConnectionStr, ok := ctx.Config[splunkVerifyConnectionKey]; ok {
		var err error
		verifyConnection, err = strconv.ParseBool(verifyConnectionStr)
		if err != nil {
			return nil, err
		}
	}
	if verifyConnection {
		err = verifySplunkConnection(logger)
		if err != nil {
			return nil, err
		}
	}

	var splunkFormat string
	if splunkFormatParsed, ok := ctx.Config[splunkFormatKey]; ok {
		switch splunkFormatParsed {
		case splunkFormatInline:
		case splunkFormatJSON:
//.........这里部分代码省略.........

// New creates splunk logger driver using configuration passed in context
func New(ctx logger.Context) (logger.Logger, error) {
	hostname, err := ctx.Hostname()
	if err != nil {
		return nil, fmt.Errorf("%s: cannot access hostname to set source field", driverName)
	}

	// Parse and validate Splunk URL
	splunkURL, err := parseURL(ctx)
	if err != nil {
		return nil, err
	}

	// Splunk Token is required parameter
	splunkToken, ok := ctx.Config[splunkTokenKey]
	if !ok {
		return nil, fmt.Errorf("%s: %s is expected", driverName, splunkTokenKey)
	}

	tlsConfig := &tls.Config{}

	// Splunk is using autogenerated certificates by default,
	// allow users to trust them with skipping verification
	if insecureSkipVerifyStr, ok := ctx.Config[splunkInsecureSkipVerifyKey]; ok {
		insecureSkipVerify, err := strconv.ParseBool(insecureSkipVerifyStr)
		if err != nil {
			return nil, err
		}
		tlsConfig.InsecureSkipVerify = insecureSkipVerify
	}

	// If path to the root certificate is provided - load it
	if caPath, ok := ctx.Config[splunkCAPathKey]; ok {
		caCert, err := ioutil.ReadFile(caPath)
		if err != nil {
			return nil, err
		}
		caPool := x509.NewCertPool()
		caPool.AppendCertsFromPEM(caCert)
		tlsConfig.RootCAs = caPool
	}

	if caName, ok := ctx.Config[splunkCANameKey]; ok {
		tlsConfig.ServerName = caName
	}

	transport := &http.Transport{
		TLSClientConfig: tlsConfig,
	}
	client := &http.Client{
		Transport: transport,
	}

	var nullMessage = &splunkMessage{
		Host: hostname,
	}

	// Optional parameters for messages
	nullMessage.Source = ctx.Config[splunkSourceKey]
	nullMessage.SourceType = ctx.Config[splunkSourceTypeKey]
	nullMessage.Index = ctx.Config[splunkIndexKey]

	tag, err := loggerutils.ParseLogTag(ctx, "{{.ID}}")
	if err != nil {
		return nil, err
	}
	nullMessage.Event.Tag = tag
	nullMessage.Event.Attrs = ctx.ExtraAttributes(nil)

	logger := &splunkLogger{
		client:      client,
		transport:   transport,
		url:         splunkURL.String(),
		auth:        "Splunk " + splunkToken,
		nullMessage: nullMessage,
	}

	err = verifySplunkConnection(logger)
	if err != nil {
		return nil, err
	}

	return logger, nil
}

// New creates a fluentd logger using the configuration passed in on
// the context. The supported context configuration variable is
// fluentd-address.
func New(ctx logger.Context) (logger.Logger, error) {
	loc, err := parseAddress(ctx.Config[addressKey])
	if err != nil {
		return nil, err
	}

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

	extra := ctx.ExtraAttributes(nil)

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

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

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

	asyncConnect := false
	if ctx.Config[asyncConnectKey] != "" {
		if asyncConnect, err = strconv.ParseBool(ctx.Config[asyncConnectKey]); err != nil {
			return nil, err
		}
	}

	fluentConfig := fluent.Config{
		FluentPort:       loc.port,
		FluentHost:       loc.host,
		FluentNetwork:    loc.protocol,
		FluentSocketPath: loc.path,
		BufferLimit:      bufferLimit,
		RetryWait:        retryWait,
		MaxRetry:         maxRetries,
		AsyncConnect:     asyncConnect,
	}

	logrus.WithField("container", ctx.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:   ctx.ContainerID,
		containerName: ctx.ContainerName,
		writer:        log,
		extra:         extra,
	}, nil
}

// Verify inline format with a not default settings for most of options
func TestInlineFormatWithNonDefaultOptions(t *testing.T) {
	hec := NewHTTPEventCollectorMock(t)

	go hec.Serve()

	ctx := logger.Context{
		Config: map[string]string{
			splunkURLKey:             hec.URL(),
			splunkTokenKey:           hec.token,
			splunkSourceKey:          "mysource",
			splunkSourceTypeKey:      "mysourcetype",
			splunkIndexKey:           "myindex",
			splunkFormatKey:          splunkFormatInline,
			splunkGzipCompressionKey: "true",
			tagKey:    "{{.ImageName}}/{{.Name}}",
			labelsKey: "a",
		},
		ContainerID:        "containeriid",
		ContainerName:      "/container_name",
		ContainerImageID:   "contaimageid",
		ContainerImageName: "container_image_name",
		ContainerLabels: map[string]string{
			"a": "b",
		},
	}

	hostname, err := ctx.Hostname()
	if err != nil {
		t.Fatal(err)
	}

	loggerDriver, err := New(ctx)
	if err != nil {
		t.Fatal(err)
	}

	if !hec.connectionVerified {
		t.Fatal("By default connection should be verified")
	}

	splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerInline)
	if !ok {
		t.Fatal("Unexpected Splunk Logging Driver type")
	}

	if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" ||
		splunkLoggerDriver.auth != "Splunk "+hec.token ||
		splunkLoggerDriver.nullMessage.Host != hostname ||
		splunkLoggerDriver.nullMessage.Source != "mysource" ||
		splunkLoggerDriver.nullMessage.SourceType != "mysourcetype" ||
		splunkLoggerDriver.nullMessage.Index != "myindex" ||
		splunkLoggerDriver.gzipCompression != true ||
		splunkLoggerDriver.gzipCompressionLevel != gzip.DefaultCompression ||
		splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency ||
		splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize ||
		splunkLoggerDriver.bufferMaximum != defaultBufferMaximum ||
		cap(splunkLoggerDriver.stream) != defaultStreamChannelSize {
		t.Fatal("Values do not match configuration.")
	}

	messageTime := time.Now()
	if err := loggerDriver.Log(&logger.Message{[]byte("1"), "stdout", messageTime, nil, false}); err != nil {
		t.Fatal(err)
	}

	err = loggerDriver.Close()
	if err != nil {
		t.Fatal(err)
	}

	if len(hec.messages) != 1 {
		t.Fatal("Expected on