// Configure initializes this producer with values from a plugin config.
190580
func (prod *Console) Configure(conf core.PluginConfigReader) {
	console := conf.GetString("Console", "stdout")

	switch strings.ToLower(console) {
	default:
		fallthrough
	case "stdout":
		prod.console = os.Stdout
	case "stderr":
		prod.console = os.Stderr
	}
}
685764
func (w *unixConfigure) Configure() error {
	if err := w.ConfigureBasic(); err != nil {
		return err
	}
	if err := w.ConfigureJuju(); err != nil {
		return err
	}
	return w.ConfigureCustomOverrides()
}
475684
func (format *MetadataCopy) Configure(conf core.PluginConfigReader) {
	mode := conf.GetString("Mode", "replace")
	switch strings.ToLower(mode) {
	case "replace":
		format.mode = metadataCopyModeReplace
	case "append":
		format.mode = metadataCopyModeAppend
	case "prepend":
		format.mode = metadataCopyModePrepend
	default:
		conf.Errors.Pushf("mode must be one of replace, append or prepend")
	}
}
965564
func (prod *KafkaProducer) Configure(conf core.PluginConfigReader) error {
	prod.SetStopCallback(prod.close)

	prod.streamToTopic = conf.GetStreamMap("Topic", "default")
	prod.topic = make(map[core.MessageStreamID]*topicHandle)

	prod.topicGuard = new(sync.RWMutex)
	prod.streamToTopic = conf.GetStreamMap("Topic", "default")

	prod.topicHandles = make(map[string]*topicHandle)

	// Init librdkafka
	prod.config = kafka.NewConfig()

	/*kafkaVer := conf.GetString("ProtocolVersion", "0.8.2")
	verParts := strings.Split(kafkaVer, ".")
	multiplicator := 1000000
	verNumber := 0
	for _, n := range verParts {
		part, _ := strconv.Atoi(n)
		verNumber += part * multiplicator
		multiplicator /= 100
	}*/

	prod.config.Set("metadata.broker.list", strings.Join(prod.servers, ","))
	prod.config.Set("sasl.mechanisms", conf.GetString("SaslMechanism", ""))
	prod.config.Set("sasl.password", conf.GetString("SaslPassword", ""))
	prod.config.Set("sasl.username", conf.GetString("SaslUsername", ""))
	prod.config.Set("ssl.ca.location", conf.GetString("SslCaLocation", ""))
	prod.config.Set("ssl.certificate.location", conf.GetString("SslCertificateLocation", ""))
	prod.config.Set("ssl.cipher.suites", conf.GetString("SslCipherSuites", ""))
	prod.config.Set("ssl.crl.location", conf.GetString("SslCrlLocation", ""))
	prod.config.Set("ssl.key.location", conf.GetString("SslKeyLocation", ""))
	prod.config.Set("ssl.key.password", conf.GetString("SslKeyPassword", ""))
	prod.config.SetI("message.max.bytes", int(conf.GetInt("Batch/SizeMaxKB", 1<<10))<<10)
	prod.config.SetI("metadata.request.timeout.ms", int(conf.GetInt("MetadataTimeoutMs", 1500)))
	prod.config.SetI("topic.metadata.refresh.interval.ms", int(conf.GetInt("MetadataRefreshMs", 300000)))
	prod.config.SetI("socket.max.fails", int(conf.GetInt("ServerMaxFails", 3)))
	prod.config.SetI("socket.timeout.ms", int(conf.GetInt("ServerTimeoutSec", 60)*1000))
	prod.config.SetB("socket.keepalive.enable", true)
	prod.config.SetI("message.send.max.retries", int(conf.GetInt("SendRetries", 0)))
	prod.config.SetI("queue.buffering.max.messages", int(conf.GetInt("Batch/MaxMessages", 100000)))
	prod.config.SetI("queue.buffering.max.ms", int(conf.GetInt("Batch/TimeoutMs", 1000)))
	prod.config.SetI("batch.num.messages", int(conf.GetInt("Batch/MinMessages", 1000)))
	//prod.config.SetI("protocol.version", verNumber)

	securityProtocol := strings.ToLower(conf.GetString("SecurityProtocol", protocolPlaintext))
	switch securityProtocol {
	default:
		return fmt.Errorf("%s is not a valid security protocol", securityProtocol)
	case protocolPlaintext:
		prod.config.Set("security.protocol", protocolPlaintext)
	case protocolSsl:
		prod.config.Set("security.protocol", protocolSsl)
	case protocolSaslPlaintext:
		prod.config.Set("security.protocol", protocolSaslPlaintext)
	case protocolSaslSsl:
		prod.config.Set("security.protocol", protocolSaslSsl)
	}

	switch strings.ToLower(conf.GetString("Compression", compressNone)) {
	default:
		prod.config.Set("compression.codec", "none")
	case compressGZIP:
		prod.config.Set("compression.codec", "gzip")
	case compressSnappy:
		prod.config.Set("compression.codec", "snappy")
	}

	tgo.Metric.New(kafkaMetricAllocations)
	return nil
}
338023
func Configure(options Options) (io.Closer, error) {
	cfg := config.Configuration{
		Disabled: options.Disabled,
		Sampler: &config.SamplerConfig{
			Type:  jaeger.SamplerTypeProbabilistic,
			Param: options.Probability,
		},
	}

	return cfg.InitGlobalTracer(options.ServiceName)
}
710743
func (format *CollectdToInflux10) Configure(conf core.PluginConfigReader) {
	format.tagString = strings.NewReplacer(",", "\\,", " ", "\\ ")
}
1272712
func (prod *Scribe) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)
	host := conf.GetString("Address", "localhost:1463")
	prod.category = conf.GetStreamMap("Categories", "")
	prod.windowSize = prod.maxWindowSize
	prod.categoryGuard = new(sync.RWMutex)
	prod.metricsRegistry = core.NewMetricsRegistryForPlugin(prod)
	prod.metricCount = make(map[core.MessageStreamID]metrics.Counter)

	// Initialize scribe connection

	var err error
	prod.socket, err = thrift.NewTSocketTimeout(host, prod.connectionTimeout)
	conf.Errors.Push(err)

	prod.transport = thrift.NewTFramedTransport(prod.socket)
	binProtocol := thrift.NewTBinaryProtocol(prod.transport, false, false)
	prod.scribe = scribe.NewScribeClientProtocol(prod.transport, binProtocol, binProtocol)

	prod.metricWindowSize = metrics.NewGauge()
	prod.metricsRegistry.Register("windowsize", prod.metricWindowSize)

	for streamID, category := range prod.category {
		counter := metrics.NewCounter()
		prod.metricCount[streamID] = counter
		prod.metricsRegistry.Register(category, counter)
	}
}
1091916
func (prod *AwsFirehose) Configure(conf core.PluginConfigReader) {
	prod.lastSendTime = time.Now()
	prod.metricsRegistry = core.NewMetricsRegistryForPlugin(prod)
	prod.metricCount = make(map[string]metrics.Counter)

	if prod.recordMaxMessages < 1 {
		prod.recordMaxMessages = 1
		prod.Logger.Warning("RecordMaxMessages was < 1. Defaulting to 1.")
	}

	if prod.recordMaxMessages > 1 && len(prod.delimiter) == 0 {
		prod.delimiter = []byte("\n")
		prod.Logger.Warning("RecordMessageDelimiter was empty. Defaulting to \"\\n\".")
	}

	prod.streamMap = conf.GetStreamMap("StreamMapping", "default")
	for _, firehoseStreamName := range prod.streamMap {
		counter := metrics.NewCounter()
		prod.metricCount[firehoseStreamName] = counter
		prod.metricsRegistry.Register(firehoseStreamName, counter)
	}
}
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)
	}
}
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)
	}
}
777728
func (prod *Redis) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	prod.protocol, prod.address = tnet.ParseAddress(conf.GetString("Address", ":6379"), "tcp")

	switch strings.ToLower(conf.GetString("Storage", "hash")) {
	case "hash":
		prod.store = prod.storeHash
	case "list":
		prod.store = prod.storeList
	case "set":
		prod.store = prod.storeSet
	case "sortedset":
		prod.store = prod.storeSortedSet
	default: // string
		prod.store = prod.storeString
	}
}
1312073
func (cons *SimpleConsumer) Configure(conf PluginConfigReader) {
	cons.id = conf.GetID()
	cons.Logger = conf.GetLogger()
	cons.runState = NewPluginRunState()
	cons.control = make(chan PluginControl, 1)

	numRoutines := conf.GetInt("ModulatorRoutines", 0)
	queueSize := conf.GetInt("ModulatorQueueSize", 1024)

	if numRoutines > 0 {
		cons.Logger.Debugf("Using %d modulator routines", numRoutines)
		cons.modulatorQueue = NewMessageQueue(int(queueSize))
		for i := 0; i < int(numRoutines); i++ {
			go cons.processQueue()
		}
		cons.enqueueMessage = cons.parallelEnqueue
	} else {
		cons.enqueueMessage = cons.directEnqueue
	}

	// Simple health check for the plugin state
	//   Path: "/<plugin_id>/pluginState"
	cons.AddHealthCheckAt("/pluginState", func() (code int, body string) {
		if cons.IsActive() {
			return thealthcheck.StatusOK, fmt.Sprintf("ACTIVE: %s", cons.runState.GetStateString())
		}
		return thealthcheck.StatusServiceUnavailable,
			fmt.Sprintf("NOT_ACTIVE: %s", cons.runState.GetStateString())
	})
}
447807
func (prod *S3) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	prod.streamMap = conf.GetStreamMap("StreamMapping", "default")
	prod.batch = core.NewMessageBatch(int(conf.GetInt("BatchMaxMessages", 5000)))

	prod.lastSendTime = time.Now()
	prod.closing = false
	prod.objects = make(map[string]*objectData)
	prod.objectsLock = new(sync.Mutex)

	prod.metricsRegistry = core.NewMetricsRegistry(prod.GetID())
	prod.counters = make(map[string]metrics.Counter)

	for _, s3Path := range prod.streamMap {
		counter := metrics.NewCounter()
		prod.counters[s3Path] = counter
		prod.metricsRegistry.Register(s3Path, counter)
	}

	prod.useFiles = prod.localPath != ""
	if prod.useFiles {
		if err := os.MkdirAll(prod.localPath, 0700); err != nil {
			conf.Errors.Pushf("Failed to create %s because of %s", prod.localPath, err.Error())
			return // ### return, missing directory ###
		}
		prod.stateFile = path.Join(prod.localPath, "state")
		if _, err := os.Stat(prod.stateFile); !os.IsNotExist(err) {
			data, err := ioutil.ReadFile(prod.stateFile)
			if conf.Errors.Push(err) {
				return
			}
			if err := json.Unmarshal(data, &prod.objects); conf.Errors.Push(err) {
				return
			}
			for s3Path, object := range prod.objects {
				filename := object.Filename
				basename := path.Base(filename)
				if basenum, err := strconv.ParseInt(basename, 10, 64); err == nil && basenum > prod.nextFile {
					prod.nextFile = basenum
				}
				if object.Compressed {
					filename += ".gz"
				}
				buffer, err := newS3FileBuffer(filename,
					prod.Logger.WithField("Scope", "fileBuffer"))
				if conf.Errors.Push(err) {
					return
				}
				object.buffer = buffer
				object.lock = new(sync.Mutex)
				// add missing metrics
				if _, exists := prod.counters[s3Path]; !exists {
					counter := metrics.NewCounter()
					prod.counters[s3Path] = counter
					prod.metricsRegistry.GetOrRegister(s3Path, counter)
				}
			}
		}
	}

	if conf.HasValue("PathFormatter") {
		keyFormatter := conf.GetPlugin("PathFormatter", "format.Identifier", tcontainer.MarshalMap{})
		prod.pathFormat = keyFormatter.(core.Modulator)
	} else {
		prod.pathFormat = nil
	}

	if prod.objectMaxMessages < 1 {
		prod.objectMaxMessages = 1
		prod.Logger.Warning("ObjectMaxMessages was < 1. Defaulting to 1.")
	}

	if prod.objectMaxMessages > 1 && len(prod.delimiter) == 0 {
		prod.delimiter = []byte("\n")
		prod.Logger.Warning("ObjectMessageDelimiter was empty. Defaulting to \"\\n\".")
	}

	// Config
	defaultEndpoint := "s3.amazonaws.com"
	prod.config = aws.NewConfig()
	if region := conf.GetString("Region", "eu-west-1"); region != "" {
		prod.config.WithRegion(region)
		if region != "us-east-1" {
			defaultEndpoint = "s3-" + region + ".amazonaws.com"
		}
	}

	if endpoint := conf.GetString("Endpoint", defaultEndpoint); endpoint != "" {
		prod.config.WithEndpoint(endpoint)
	}

	// Credentials
	prod.config.CredentialsChainVerboseErrors = aws.Bool(true)
	credentialType := strings.ToLower(conf.GetString("Credential/Type", s3CredentialNone))
	switch credentialType {
	case s3CredentialEnv:
		prod.config.WithCredentials(credentials.NewEnvCredentials())

	case s3CredentialStatic:
		id := conf.GetString("Credential/Id", "")
		token := conf.GetString("Credential/Token", "")
		secret := conf.GetString("Credential/Secret", "")
		prod.config.WithCredentials(credentials.NewStaticCredentials(id, secret, token))

	case s3CredentialShared:
		filename := conf.GetString("Credential/File", "")
		profile := conf.GetString("Credential/Profile", "")
		prod.config.WithCredentials(credentials.NewSharedCredentials(filename, profile))

	case s3CredentialNone:
		// Nothing

	default:
		conf.Errors.Pushf("Unknown CredentialType: %s", credentialType)
	}
}
668874
func (client *AwsMultiClient) Configure(conf core.PluginConfigReader) {
	client.config = aws.NewConfig()

	client.config.WithRegion(client.region)
	client.config.WithEndpoint(client.endpoint)

	credentials, err := client.Credentials.CreateAwsCredentials()
	if err != nil {
		conf.Errors.Push(err)
	}

	client.config.WithCredentials(credentials)
	client.config.CredentialsChainVerboseErrors = aws.Bool(true)
}
1273916
func (prod *Proxy) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	prod.protocol, prod.address = tnet.ParseAddress(conf.GetString("Address", ":5880"), "tcp")
	if prod.protocol == "udp" {
		conf.Errors.Pushf("Proxy does not support UDP")
	}

	delimiter := tstrings.Unescape(conf.GetString("Delimiter", "\n"))
	offset := int(conf.GetInt("Offset", 0))
	flags := tio.BufferedReaderFlagEverything // pass all messages as-is

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

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

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

	case "ascii":
		flags |= tio.BufferedReaderFlagMLE

	case "delimiter":
		// Nothing to add

	default:
		conf.Errors.Pushf("Unknown partitioner: %s", partitioner)
	}

	prod.reader = tio.NewBufferedReader(prod.bufferSizeKB, flags, offset, delimiter)
}
446502
func (prod *AwsKinesis) Configure(conf core.PluginConfigReader) {
	prod.lastSendTime = time.Now()
	prod.counters = make(map[string]*int64)
	prod.lastMetricUpdate = time.Now()
	prod.sequence = new(int64)
	prod.metricsRegistry = core.NewMetricsRegistryForPlugin(prod)
	prod.metricCount = make(map[string]metrics.Counter)

	if prod.recordMaxMessages < 1 {
		prod.recordMaxMessages = 1
		prod.Logger.Warning("RecordMaxMessages was < 1. Defaulting to 1.")
	}

	if prod.recordMaxMessages > 1 && len(prod.delimiter) == 0 {
		prod.delimiter = []byte("\n")
		prod.Logger.Warning("RecordMessageDelimiter was empty. Defaulting to \"\\n\".")
	}

	prod.streamMap = conf.GetStreamMap("StreamMapping", "")
	for _, kinesisStreamName := range prod.streamMap {
		counter := metrics.NewCounter()
		prod.metricCount[kinesisStreamName] = counter
		prod.metricsRegistry.Register(kinesisStreamName, counter)
	}
}
308086
func (cons *File) Configure(conf core.PluginConfigReader) {
	cons.done = make(chan struct{})
	cons.observedFiles = new(sync.Map)

	// TODO: support manual roll again
	//cons.SetRollCallback(cons.onRoll)
	cons.SetStopCallback(func() {
		close(cons.done)
	})

	// restore default observer mode for invalid config settings
	if cons.observeMode != observeModePoll && cons.observeMode != observeModeWatch {
		cons.Logger.Warningf("Unknown observe mode '%s'. Using poll", cons.observeMode)
		cons.observeMode = observeModePoll
	}

	cons.configureBlacklist(conf)
}