// Configure initializes this filter with values from a plugin config.
196649
func (filter *Sample) Configure(conf core.PluginConfigReader) {
	filter.count = new(uint64)
	filter.ignore = make(map[core.MessageStreamID]bool)
	ignore := conf.GetStreamArray("SampleIgnore", []core.MessageStreamID{})
	for _, stream := range ignore {
		filter.ignore[stream] = true
	}
}
735878
func (b *Bootstrapper) Configure(cs ...Configurator) {
	for _, c := range cs {
		c(b)
	}
}
235341
func (prod *Spooling) Configure(conf core.PluginConfigReader) {
	prod.SetPrepareStopCallback(prod.waitForReader)
	prod.SetStopCallback(prod.close)
	prod.SetRollCallback(prod.onRoll)
	prod.metricsRegistry = core.NewMetricsRegistryForPlugin(prod)

	prod.outfileGuard = new(sync.RWMutex)
	prod.outfile = make(map[core.MessageStreamID]*spoolFile)

	if maxMsgSec := time.Duration(conf.GetInt("MaxMessagesSec", 100)); maxMsgSec > 0 {
		prod.readDelay = time.Second / maxMsgSec
	} else {
		prod.readDelay = 0
	}

	//TODO: check if rotation is still in use
	prod.rotation = components.NewRotateConfig()
	prod.rotation.Enabled = true
	prod.rotation.Timeout = prod.maxFileAge
	prod.rotation.SizeByte = prod.maxFileSize
}
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")
	}
}
1224428
func (app *Application) Configure(configurators ...func(*Application)) *Application {
	for _, c := range configurators {
		c(app)
	}
	return app
}
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)
	}
}
766214
func (format *JSONToInflux10) Configure(conf core.PluginConfigReader) {
	format.SimpleFormatter.Configure(conf)

	format.metadataEscape = strings.NewReplacer(",", "\\,", "=", "\\=", " ", "\\ ")
	format.measurementEscape = strings.NewReplacer(",", "\\,", " ", "\\ ")
	format.fieldValueEscape = strings.NewReplacer("\"", "\\\"")

	// Create a set from the tags array for O(1) lookup
	tagsArray := conf.GetStringArray("Tags", []string{})
	format.tags = map[string]bool{}
	for _, v := range tagsArray {
		format.tags[v] = true
	}

	// Create a set from the ignore array for O(1) lookup
	ignoreArray := conf.GetStringArray("Ignore", []string{})
	format.ignore = map[string]bool{}
	for _, v := range ignoreArray {
		format.ignore[v] = true
	}
}
1362698
func Configure(apiServer, kubeconfigPath string) {
	config.APIServer = apiServer
	config.KubeconfigPath = kubeconfigPath

	if IsEnabled() &&
		config.APIServer != "" &&
		!strings.HasPrefix(apiServer, "http") {
		config.APIServer = "http://" + apiServer
	}
}
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)
}
1008001
func (cons *Syslogd) Configure(conf core.PluginConfigReader) {
	cons.protocol, cons.address = tnet.ParseAddress(
		conf.GetString("Address", "udp://0.0.0.0:514"), "tcp")

	syslogFormat := conf.GetString("Format", "RFC6587")

	switch cons.protocol {
	case "udp", "tcp", "unix":
	default:
		conf.Errors.Pushf("Unknown protocol type %s", cons.protocol) // ### return, unknown protocol ###
	}

	switch syslogFormat {
	// http://www.ietf.org/rfc/rfc3164.txt
	case "RFC3164":
		cons.format = syslog.RFC3164
		if cons.protocol != "udp" {
			cons.Logger.Infof("Using '%s' instead of 'udp' for RFC3164 violates the standard", cons.protocol)
		}

	// https://tools.ietf.org/html/rfc5424
	case "RFC5424":
		cons.format = syslog.RFC5424
		if cons.protocol != "udp" {
			cons.Logger.Infof("Using '%s' instead of 'udp' for RFC5424 violates the standard", cons.protocol)
		}

	// https://tools.ietf.org/html/rfc6587
	case "RFC6587":
		cons.format = syslog.RFC6587

	default:
		conf.Errors.Pushf("Format %s is not supported", syslogFormat)
	}
}
397264
func (prod *Kafka) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	kafka.Logger = prod.Logger.WithField("Scope", "Sarama")

	prod.topicGuard = new(sync.RWMutex)
	prod.streamToTopic = conf.GetStreamMap("Topics", "")
	prod.topic = make(map[core.MessageStreamID]*topicHandle)
	prod.topicHandles = make(map[string]*topicHandle)
	prod.metricsRegistry = core.NewMetricsRegistryForPlugin(prod)

	prod.config = kafka.NewConfig()
	prod.config.ClientID = prod.clientID
	prod.config.ChannelBufferSize = int(conf.GetInt("MessageBufferCount", 8192))

	switch ver := conf.GetString("Version", "0.8.2"); ver {
	case "0.8.2.0":
		prod.config.Version = kafka.V0_8_2_0
	case "0.8.2.1":
		prod.config.Version = kafka.V0_8_2_1
	case "0.8", "0.8.2", "0.8.2.2":
		prod.config.Version = kafka.V0_8_2_2
	case "0.9.0", "0.9.0.0":
		prod.config.Version = kafka.V0_9_0_0
	case "0.9", "0.9.0.1":
		prod.config.Version = kafka.V0_9_0_1
	case "0.10", "0.10.0", "0.10.0.0":
		prod.config.Version = kafka.V0_10_0_0
	default:
		prod.Logger.Warning("Unknown kafka version given: ", ver)
		parts := strings.Split(ver, ".")
		if len(parts) < 2 {
			prod.config.Version = kafka.V0_8_2_2
		} else {
			minor, _ := strconv.ParseUint(parts[1], 10, 8)
			switch {
			case minor <= 8:
				prod.config.Version = kafka.V0_8_2_2
			case minor == 9:
				prod.config.Version = kafka.V0_9_0_1
			case minor >= 10:
				prod.config.Version = kafka.V0_10_0_0
			}
		}
	}

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

	prod.config.Net.TLS.Enable = conf.GetBool("TlsEnable", false)
	if prod.config.Net.TLS.Enable {
		prod.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
			}
			prod.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)
		prod.config.Net.TLS.Config.RootCAs = caCertPool

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

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

	}

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

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

	prod.config.Producer.RequiredAcks = kafka.RequiredAcks(conf.GetInt("RequiredAcks", int64(kafka.WaitForLocal)))
	prod.config.Producer.Timeout = time.Duration(conf.GetInt("TimeoutMs", 10000)) * time.Millisecond
	prod.config.Producer.MaxMessageBytes = int(conf.GetInt("Batch/SizeMaxKB", 1<<10)) << 10
	prod.config.Producer.Flush.Bytes = int(conf.GetInt("Batch/MinSizeByte", 8192))
	prod.config.Producer.Flush.Messages = int(conf.GetInt("Batch/MinCount", 1))
	prod.config.Producer.Flush.Frequency = time.Duration(conf.GetInt("Batch/TimeoutMs", 3000)) * time.Millisecond
	prod.config.Producer.Flush.MaxMessages = int(conf.GetInt("Batch/MaxCount", 0))
	prod.config.Producer.Retry.Max = int(conf.GetInt("SendRetries", 1))
	prod.config.Producer.Retry.Backoff = time.Duration(conf.GetInt("SendTimeoutMs", 100)) * time.Millisecond

	prod.config.Producer.Return.Successes = true
	prod.config.Producer.Return.Errors = true
	prod.config.Producer.Return.Successes = true

	switch strings.ToLower(conf.GetString("Compression", compressNone)) {
	default:
		fallthrough
	case compressNone:
		prod.config.Producer.Compression = kafka.CompressionNone
	case compressGZIP:
		prod.config.Producer.Compression = kafka.CompressionGZIP
	case compressSnappy:
		prod.config.Producer.Compression = kafka.CompressionSnappy
	}

	switch strings.ToLower(conf.GetString("Partitioner", partRoundrobin)) {
	case partRandom:
		prod.config.Producer.Partitioner = kafka.NewRandomPartitioner
	case partRoundrobin:
		prod.config.Producer.Partitioner = kafka.NewRoundRobinPartitioner
	case partHash:
		fallthrough
	default:
		switch strings.ToLower(conf.GetString("PartitionHasher", "FNV-1a")) {
		case "murmur2":
			prod.config.Producer.Partitioner = NewMurmur2HashPartitioner
		case "fnv-1a":
			fallthrough
		default:
			prod.config.Producer.Partitioner = kafka.NewHashPartitioner

		}
	}
}
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)
}
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())
}
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)
}
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())
	})
}
1175117
func (filter *Rate) Configure(conf core.PluginConfigReader) {
	filter.stateGuard = new(sync.RWMutex)
	filter.state = make(map[core.MessageStreamID]*rateState)
	filter.metricsRegistry = core.NewMetricsRegistry("ratelimit")

	ignore := conf.GetStreamArray("Ignore", []core.MessageStreamID{})
	for _, stream := range ignore {
		filter.state[stream] = &rateState{
			ignore: true,
		}
	}
}