// Configure initializes this consumer with values from a plugin config.
484071
func (cons *Socket) Configure(conf core.PluginConfigReader) {
	if len(cons.acknowledge) > 0 && cons.protocol == "udp" {
		conf.Errors.Pushf("UDP sockets do not support acknowledgment.")
	}

	address := conf.GetString("Address", "tcp://0.0.0.0:5880")
	cons.protocol, cons.address = tnet.ParseAddress(address, "tcp")
	cons.flags = 0

	partitioner := conf.GetString("Partitioner", "delimiter")
	switch strings.ToLower(partitioner) {
	case "binary_be":
		cons.flags |= tio.BufferedReaderFlagBigEndian
		fallthrough

	case "binary", "binary_le":
		cons.flags |= tio.BufferedReaderFlagEverything
		switch conf.GetInt("Size", 4) {
		case 1:
			cons.flags |= tio.BufferedReaderFlagMLE8
		case 2:
			cons.flags |= tio.BufferedReaderFlagMLE16
		case 4:
			cons.flags |= tio.BufferedReaderFlagMLE32
		case 8:
			cons.flags |= tio.BufferedReaderFlagMLE64
		default:
			conf.Errors.Pushf("Size only supports the value 1,2,4 and 8")
		}

	case "fixed":
		cons.flags |= tio.BufferedReaderFlagMLEFixed
		cons.offset = int(conf.GetInt("Size", 1))

	case "ascii":
		cons.flags |= tio.BufferedReaderFlagMLE

	case "delimiter":
		// Nothing to add

	default:
		conf.Errors.Pushf("Unknown partitioner: %s", partitioner)
	}
}
795726
func (co *Coordinator) Configure(conf *core.Config) error {
	// Make sure the log is printed to the fallback device if we are stuck here
	logFallback := time.AfterFunc(time.Duration(3)*time.Second, func() {
		logrusHookBuffer.SetTargetWriter(logger.FallbackLogDevice)
		logrusHookBuffer.Purge()
	})
	defer logFallback.Stop()

	// Initialize the plugins in the order of routers > producers > consumers
	// to match the order of reference between the different types.
	errors := tgo.NewErrorStack()
	errors.SetFormat(tgo.ErrorStackFormatCSV)

	if !co.configureRouters(conf) {
		errors.Pushf("At least one router failed to be configured")
	}
	if !co.configureProducers(conf) {
		errors.Pushf("At least one producer failed to be configured")
	}
	if !co.configureConsumers(conf) {
		errors.Pushf("At least one consumer failed to be configured")
	}
	if len(co.producers) == 0 {
		errors.Pushf("No valid producers found")
	}
	if len(co.consumers) <= 1 {
		errors.Pushf("No valid consumers found")
	}

	// As consumers might create new fallback router this is the first position
	// where we can add the wildcard producers to all streams. No new routers
	// created beyond this point must use StreamRegistry.AddWildcardProducersToRouter.

	core.StreamRegistry.AddAllWildcardProducersToAllRouters()
	return errors.OrNil()
}
773018
func (prod *StatsdMetrics) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	prod.streamMap = conf.GetStreamMap("StreamMapping", "")
	prod.batch = core.NewMessageBatch(int(conf.GetInt("Batch/MaxMessages", 500)))

	server := conf.GetString("Server", "localhost:8125")
	prefix := conf.GetString("Prefix", "gollum.")
	prod.client = statsd.NewStatsdClient(server, prefix)
	conf.Errors.Push(prod.client.CreateSocket())
}
1063412
func (cons *Proxy) Configure(conf core.PluginConfigReader) {
	cons.protocol, cons.address = tnet.ParseAddress(conf.GetString("Address", ":5880"), "tcp")
	if cons.protocol == "udp" {
		conf.Errors.Pushf("UDP is not supported")
	}

	cons.flags = tio.BufferedReaderFlagEverything

	partitioner := strings.ToLower(conf.GetString("Partitioner", "delimiter"))
	switch partitioner {
	case "binary_be":
		cons.flags |= tio.BufferedReaderFlagBigEndian
		fallthrough

	case "binary", "binary_le":
		switch cons.size {
		case 1:
			cons.flags |= tio.BufferedReaderFlagMLE8
		case 2:
			cons.flags |= tio.BufferedReaderFlagMLE16
		case 4:
			cons.flags |= tio.BufferedReaderFlagMLE32
		case 8:
			cons.flags |= tio.BufferedReaderFlagMLE64
		default:
			conf.Errors.Pushf("Size only supports the value 1,2,4 and 8")
		}

	case "fixed":
		cons.flags |= tio.BufferedReaderFlagMLEFixed
		cons.offset = cons.size

	case "ascii":
		cons.flags |= tio.BufferedReaderFlagMLE

	case "delimiter":
		// Nothing to add

	default:
		conf.Errors.Pushf("Unknown partitioner: %s", partitioner)
	}
}
1138199
func (sqlite3 *Sqlite3DB) Configure(cfg map[string]string) {
	sqlite3.DBPath = cfg["DBPath"]
	sqlite3.TableName = cfg["TableName"]
}
209838
func (r *Router) Configure(opts ...RouterOption) {
	for _, o := range opts {
		o(r)
	}
}
1228799
func (format *TemplateJSON) Configure(conf core.PluginConfigReader) {
	var err error
	tpl := conf.GetString("Template", "")
	format.template, err = template.New("Template").Parse(tpl)
	conf.Errors.Push(err)
}
735878
func (b *Bootstrapper) Configure(cs ...Configurator) {
	for _, c := range cs {
		c(b)
	}
}
1177384
func (p *PostProcessor) Configure(raws ...interface{}) error {
	err := config.Decode(&p.config, &config.DecodeOpts{
		Interpolate:        true,
		InterpolateContext: &p.config.ctx,
		InterpolateFilter: &interpolate.RenderFilter{
			Exclude: []string{
				"oss_key_name",
			},
		},
	}, raws...)
	if err != nil {
		return err
	}

	errs := new(packer.MultiError)

	// Check and render oss_key_name
	if err = interpolate.Validate(p.config.OSSKey, &p.config.ctx); err != nil {
		errs = packer.MultiErrorAppend(
			errs, fmt.Errorf("Error parsing oss_key_name template: %s", err))
	}

	// Check we have alicloud access variables defined somewhere
	errs = packer.MultiErrorAppend(errs, p.config.AlicloudAccessConfig.Prepare(&p.config.ctx)...)

	// define all our required parameters
	templates := map[string]*string{
		"oss_bucket_name": &p.config.OSSBucket,
	}
	// Check out required params are defined
	for key, ptr := range templates {
		if *ptr == "" {
			errs = packer.MultiErrorAppend(
				errs, fmt.Errorf("%s must be set", key))
		}
	}

	// Anything which flagged return back up the stack
	if len(errs.Errors) > 0 {
		return errs
	}

	packer.LogSecretFilter.Set(p.config.AlicloudAccessKey, p.config.AlicloudSecretKey)
	log.Println(p.config)
	return nil
}
908056
func (router *SimpleRouter) Configure(conf PluginConfigReader) {
	router.id = conf.GetID()
	router.Logger = conf.GetLogger()

	if router.streamID == WildcardStreamID && strings.Index(router.id, GeneratedRouterPrefix) != 0 {
		router.Logger.Info("A wildcard stream configuration only affects the wildcard stream, not all routers")
	}
}
519234
func (format *TextToJSON) Configure(conf core.PluginConfigReader) {
	format.parser = tstrings.NewTransitionParser()
	format.state = jsonReadObject
	format.timeParse = time.Parse
	format.parseLock = new(sync.Mutex)

	unixRead := conf.GetString("UnixTimestampRead", "")
	if format.timeRead == "" {
		if unixRead == "" {
			// Use default when neither are specified
			format.timeRead = "20060102150405"
		} else {
			format.timeRead = unixRead
			format.timeParse = parseUnix
		}
	} else if unixRead != "" {
		format.Logger.Warning("Cannot use both TimestampRead and UnixTimestampRead, defaulting to TimestampRead")
	}

	if !conf.HasValue("Directives") {
		format.Logger.Warning("JSON formatter has no directives setting")
		return // ### return, no directives ###
	}

	directiveStrings := conf.GetStringArray("Directives", []string{})
	if len(directiveStrings) == 0 {
		conf.Errors.Pushf("JSON formatter has no directives")
	} else {
		// Parse directives
		parserFunctions := make(map[string]tstrings.ParsedFunc)
		parserFunctions["key"] = format.readKey
		parserFunctions["val"] = format.readValue
		parserFunctions["esc"] = format.readEscaped
		parserFunctions["dat"] = format.readDate
		parserFunctions["arr"] = format.readArray
		parserFunctions["obj"] = format.readObject
		parserFunctions["end"] = format.readEnd
		parserFunctions["arr+val"] = format.readArrayValue
		parserFunctions["arr+dat"] = format.readArrayDate
		parserFunctions["arr+esc"] = format.readArrayEscaped
		parserFunctions["val+end"] = format.readValueEnd
		parserFunctions["esc+end"] = format.readEscapedEnd
		parserFunctions["dat+end"] = format.readDateEnd

		directives := []tstrings.TransitionDirective{}
		for _, dirString := range directiveStrings {
			directive, err := tstrings.ParseTransitionDirective(dirString, parserFunctions)
			if err != nil {
				conf.Errors.Pushf("%s: %s", err.Error(), dirString)
				continue // ### continue, malformed directive ###
			}

			if format.initState == "" {
				format.initState = directive.State
			}
			directives = append(directives, directive)
		}

		format.parser.AddDirectives(directives)

		// Validate initstate
		initStateValid := false
		for i := 0; i < len(directives) && !initStateValid; i++ {
			initStateValid = directives[i].State == format.initState
		}
		if !initStateValid {
			conf.Errors.Pushf("JSONStartState does not exist in directives")
		}

		for _, dir := range directives {
			nextStateValid := false
			for i := 0; i < len(directives) && !nextStateValid; i++ {
				nextStateValid = dir.NextState == directives[i].State
			}
			if !nextStateValid {
				format.Logger.Warningf("State \"%s\" has a transition to \"%s\" which does not exist in directives", dir.State, dir.NextState)
			}
		}
	}
}
1057469
func (prod *AwsS3) Configure(conf core.PluginConfigReader) {
	prod.SetRollCallback(prod.rotateTargetFiles)
	prod.SetStopCallback(prod.close)

	prod.filesByStream = make(map[core.MessageStreamID]*components.BatchedWriterAssembly)
	prod.files = make(map[string]*components.BatchedWriterAssembly)

	prod.hasWildcard = strings.IndexByte(prod.fileNamePattern, '*') != -1
	prod.Rotate.Enabled = true // force rotation

	prod.batchedFileGuard = new(sync.RWMutex)
}
277163
func (cons *Kafka) Configure(conf core.PluginConfigReader) {
	cons.offsets = make(map[int32]*int64)
	cons.MaxPartitionID = 0

	cons.config = kafka.NewConfig()
	cons.config.ClientID = conf.GetString("ClientId", "gollum")
	cons.config.ChannelBufferSize = int(conf.GetInt("MessageBufferCount", 8192))

	switch ver := conf.GetString("Version", "0.8.2"); ver {
	case "0.8.2.0":
		cons.config.Version = kafka.V0_8_2_0
	case "0.8.2.1":
		cons.config.Version = kafka.V0_8_2_1
	case "0.8", "0.8.2", "0.8.2.2":
		cons.config.Version = kafka.V0_8_2_2
	case "0.9.0", "0.9.0.0":
		cons.config.Version = kafka.V0_9_0_0
	case "0.9", "0.9.0.1":
		cons.config.Version = kafka.V0_9_0_1
	case "0.10", "0.10.0", "0.10.0.0":
		cons.config.Version = kafka.V0_10_0_0
	case "0.10.0.1":
		cons.config.Version = kafka.V0_10_0_1
	case "0.10.1", "0.10.1.0":
		cons.config.Version = kafka.V0_10_1_0
	case "0.10.2", "0.10.2.0":
		cons.config.Version = kafka.V0_10_2_0
	case "0.11", "0.11.0", "0.11.0.0":
		cons.config.Version = kafka.V0_11_0_0
	case "1", "1.0", "1.0.0", "1.0.0.0":
		cons.config.Version = kafka.V1_0_0_0
	default:
		cons.Logger.Warningf("Unknown kafka version given: %s. Falling back to 0.8.2", ver)
		cons.config.Version = kafka.V0_8_2_2
	}

	cons.config.Net.MaxOpenRequests = int(conf.GetInt("MaxOpenRequests", 5))
	cons.config.Net.DialTimeout = time.Duration(conf.GetInt("ServerTimeoutSec", 30)) * time.Second
	cons.config.Net.ReadTimeout = cons.config.Net.DialTimeout
	cons.config.Net.WriteTimeout = cons.config.Net.DialTimeout

	cons.config.Net.TLS.Enable = conf.GetBool("TlsEnable", false)
	if cons.config.Net.TLS.Enable {
		cons.config.Net.TLS.Config = &tls.Config{}

		keyFile := conf.GetString("TlsKeyLocation", "")
		certFile := conf.GetString("TlsCertificateLocation", "")
		if keyFile != "" && certFile != "" {
			cert, err := tls.LoadX509KeyPair(certFile, keyFile)
			if conf.Errors.Push(err) {
				return
			}
			cons.config.Net.TLS.Config.Certificates = []tls.Certificate{cert}
		} else if certFile == "" {
			conf.Errors.Pushf("Cannot specify TlsKeyLocation without TlsCertificateLocation")
			return
		} else if keyFile == "" {
			conf.Errors.Pushf("Cannot specify TlsCertificateLocation without TlsKeyLocation")
			return
		}

		caFile := conf.GetString("TlsCaLocation", "")
		if caFile == "" {
			conf.Errors.Pushf("TlsEnable is set to true, but no TlsCaLocation was specified")
			return
		}

		caCert, err := ioutil.ReadFile(caFile)
		if conf.Errors.Push(err) {
			return
		}

		caCertPool := x509.NewCertPool()
		caCertPool.AppendCertsFromPEM(caCert)
		cons.config.Net.TLS.Config.RootCAs = caCertPool

		serverName := conf.GetString("TlsServerName", "")
		if serverName != "" {
			cons.config.Net.TLS.Config.ServerName = serverName
		}

		cons.config.Net.TLS.Config.InsecureSkipVerify = conf.GetBool("TlsInsecureSkipVerify", false)
	}

	cons.config.Net.SASL.Enable = conf.GetBool("SaslEnable", false)
	if cons.config.Net.SASL.Enable {
		cons.config.Net.SASL.User = conf.GetString("SaslUser", "gollum")
		cons.config.Net.SASL.Password = conf.GetString("SaslPassword", "")
	}

	cons.config.Metadata.Retry.Max = int(conf.GetInt("ElectRetries", 3))
	cons.config.Metadata.Retry.Backoff = time.Duration(conf.GetInt("ElectTimeoutMs", 250)) * time.Millisecond
	cons.config.Metadata.RefreshFrequency = time.Duration(conf.GetInt("MetadataRefreshMs", 10000)) * time.Millisecond

	cons.config.Consumer.Fetch.Min = int32(conf.GetInt("MinFetchSizeByte", 1))
	cons.config.Consumer.Fetch.Max = int32(conf.GetInt("MaxFetchSizeByte", 0))
	cons.config.Consumer.Fetch.Default = int32(conf.GetInt("DefaultFetchSizeByte", 32768))
	cons.config.Consumer.MaxWaitTime = time.Duration(conf.GetInt("FetchTimeoutMs", 250)) * time.Millisecond

	if cons.group != "" {
		cons.offsetFile = "" // forcibly ignore this option
		switch cons.config.Version {
		case kafka.V0_8_2_0, kafka.V0_8_2_1, kafka.V0_8_2_2:
			cons.Logger.Warning("Invalid kafka version 0.8.x given, minimum is 0.9 for consumer groups, defaulting to 0.9.0.1")
			cons.config.Version = kafka.V0_9_0_1
		}

		cons.groupConfig = cluster.NewConfig()
		cons.groupConfig.Config = *cons.config
	}

	offsetValue := strings.ToLower(conf.GetString("DefaultOffset", kafkaOffsetNewest))
	switch offsetValue {
	case kafkaOffsetNewest:
		cons.defaultOffset = kafka.OffsetNewest

	case kafkaOffsetOldest:
		cons.defaultOffset = kafka.OffsetOldest

	default:
		cons.defaultOffset, _ = strconv.ParseInt(offsetValue, 10, 64)
	}

	if cons.offsetFile != "" {
		fileContents, err := ioutil.ReadFile(cons.offsetFile)
		if err != nil {
			cons.Logger.Warningf("Failed to open kafka offset file: %s", err.Error())
		} else {
			// Decode the JSON file into the partition -> offset map
			encodedOffsets := make(map[string]int64)
			err = json.Unmarshal(fileContents, &encodedOffsets)
			if conf.Errors.Push(err) {
				return
			}

			for k, v := range encodedOffsets {
				id, err := strconv.Atoi(k)
				if conf.Errors.Push(err) {
					return
				}
				startOffset := v
				cons.offsets[int32(id)] = &startOffset
			}
		}
	}

	kafka.Logger = cons.Logger.WithField("Scope", "Sarama")
}
363052
func (cons *AwsKinesis) Configure(conf core.PluginConfigReader) {
	cons.offsets = make(map[string]string)
	cons.offsetsGuard = new(sync.RWMutex)

	// Offset
	offsetValue := strings.ToLower(conf.GetString("DefaultOffset", kinesisOffsetNewest))
	switch offsetValue {
	case kinesisOffsetNewest:
		cons.offsetType = kinesis.ShardIteratorTypeLatest
		cons.defaultOffset = ""

	case kinesisOffsetOldest:
		cons.offsetType = kinesis.ShardIteratorTypeTrimHorizon
		cons.defaultOffset = ""

	default:
		cons.offsetType = kinesis.ShardIteratorTypeAtSequenceNumber
		_, err := strconv.ParseUint(offsetValue, 10, 64)
		if err != nil {
			cons.Logger.Errorf("Default offset must be \"%s\", \"%s\" or a number. %s given", kinesisOffsetNewest, kinesisOffsetOldest, offsetValue)
			offsetValue = "0"
		}
		cons.defaultOffset = offsetValue
	}

	if cons.offsetFile != "" {
		fileContents, err := ioutil.ReadFile(cons.offsetFile)
		if err != nil {
			cons.Logger.Errorf("Failed to open kinesis offset file: %s", err.Error())
		} else {
			cons.offsetType = kinesis.ShardIteratorTypeAfterSequenceNumber
			conf.Errors.Push(json.Unmarshal(fileContents, &cons.offsets))
		}
	}
}
208429
func (format *SimpleFormatter) Configure(conf PluginConfigReader) {
	format.Logger = conf.GetSubLogger("Formatter")

	applyTo := conf.GetString("ApplyTo", "")
	format.GetAppliedContent = GetAppliedContentGetFunction(applyTo)
	format.SetAppliedContent = GetAppliedContentSetFunction(applyTo)
}
1073482
func (m *Module) Configure(opts ...option) {
	for _, opt := range opts {
		opt(m)
	}
}
883247
func (prod *File) Configure(conf core.PluginConfigReader) {
	prod.Pruner.Logger = prod.Logger

	prod.SetRollCallback(prod.rotateLog)
	prod.SetStopCallback(prod.close)

	prod.filesByStream = make(map[core.MessageStreamID]*components.BatchedWriterAssembly)
	prod.files = make(map[string]*components.BatchedWriterAssembly)

	logFile := conf.GetString("File", "/var/log/gollum.log")
	prod.wildcardPath = strings.IndexByte(logFile, '*') != -1

	prod.fileDir = filepath.Dir(logFile)
	prod.fileExt = filepath.Ext(logFile)
	prod.fileName = filepath.Base(logFile)
	prod.fileName = prod.fileName[:len(prod.fileName)-len(prod.fileExt)]

	prod.batchedFileGuard = new(sync.RWMutex)
}