// Configure initializes this formatter with values from a plugin config.
766293
func (format *RegExp) Configure(conf core.PluginConfigReader) {
	var err error
	if conf.GetBool("Posix", true) {
		format.expression, err = regexp.CompilePOSIX(conf.GetString("Expression", "(.*)"))
	} else {
		format.expression, err = regexp.Compile(conf.GetString("Expression", "(.*)"))
	}
	conf.Errors.Push(err)
}
697705
func (p *gateway) configure(loadedConfig *config.GatewayConfig) error {
	if loadedConfig == nil {
		return fmt.Errorf("unable to configure gateway with nil config")
	}
	// save config to the gateway
	p.config = loadedConfig

	// set debug context from loaded config
	if loadedConfig.DebugFlag != nil && *loadedConfig.DebugFlag != "" {
		p.debugContext.SetFlagStr(*loadedConfig.DebugFlag)
	}

	// set debugSink ctx flag
	p.debugSink.CtxFlagCheck = &p.debugContext

	// TODO: allow arbitrary logger to be passed into configure function
	// set lumberjack log formatting using a child derived from the default logger
	// this is used as a back up by p.getLogger when we try to create a logger using the loaded config
	p.logger = log.NewContext(log.DefaultLogger.CreateChild()).With(logkey.Time, log.DefaultTimestamp, logkey.Caller, log.DefaultCaller)

	// NOTE: the main reason to do that ^ and apply the keys is so that our log messages look consistent.
	// there is are messages logged by p.getLogger before the keys are applied in the following statement

	// create a new logger using the loaded config
	p.logger = log.NewContext(p.getLogger(loadedConfig)).With(logkey.Time, log.DefaultTimestamp, logkey.Caller, log.DefaultCaller)

	// assign logger to debug sink
	p.debugSink.Logger = p.logger

	// setup logger on versionMetric which reports our SHA1
	p.versionMetric.Logger = p.logger

	return nil
}
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())
	})
}
1073482
func (m *Module) Configure(opts ...option) {
	for _, opt := range opts {
		opt(m)
	}
}
710743
func (format *CollectdToInflux10) Configure(conf core.PluginConfigReader) {
	format.tagString = strings.NewReplacer(",", "\\,", " ", "\\ ")
}
515966
func (p *Provider) Configure(providers.ConfigureRequest) providers.ConfigureResponse {
	// At this moment there is nothing to configure for the terraform provider,
	// so we will happily return without taking any action
	var res providers.ConfigureResponse
	return res
}
879620
func (cons *LogConsumer) Configure(conf PluginConfigReader) {
	cons.control = make(chan PluginControl, 1)
	cons.logRouter = StreamRegistry.GetRouter(LogInternalStreamID)
	cons.metric = conf.GetString("MetricKey", "")
	cons.queue = NewMessageQueue(1024)

	if cons.metric != "" {
		cons.metricsRegistry = NewMetricsRegistry(cons.metric)

		cons.metricErrors = metrics.NewCounter()
		cons.metricWarning = metrics.NewCounter()
		cons.metricInfo = metrics.NewCounter()
		cons.metricDebug = metrics.NewCounter()
		cons.metricsRegistry.Register("errors", cons.metricErrors)
		cons.metricsRegistry.Register("warnings", cons.metricWarning)
		cons.metricsRegistry.Register("info", cons.metricInfo)
		cons.metricsRegistry.Register("debug", cons.metricDebug)
	}
}
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")
}
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
}
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)
}
1086355
func (format *ProcessTSV) Configure(conf core.PluginConfigReader) {
	directives := conf.GetStringArray("Directives", []string{})

	format.directives = make([]tsvDirective, 0, len(directives))
	for _, d := range directives {
		directive := strings.Replace(d, "\\:", "\r", -1)
		parts := strings.Split(directive, ":")
		for i, value := range parts {
			parts[i] = strings.Replace(value, "\r", ":", -1)
		}

		if len(parts) >= 2 {
			index, err := strconv.Atoi(parts[0])
			conf.Errors.Push(err)

			newDirective := tsvDirective{
				index:     index,
				operation: strings.ToLower(parts[1]),
			}

			for i := 2; i < len(parts); i++ {
				newDirective.parameters = append(newDirective.parameters, tstrings.Unescape(parts[i]))
			}

			format.directives = append(format.directives, newDirective)
		}
	}
}
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)
}
1081114
func (reader *PluginConfigReader) Configure(item interface{}) error {
	itemValue := reflect.ValueOf(item)
	if itemValue.Kind() != reflect.Ptr {
		panic("Configure requires reference")
	}

	var logger logrus.FieldLogger
	if loggable, isScoped := item.(ScopedLogger); isScoped {
		logger = loggable.GetLogger()
	}

	reader.configureStruct(itemValue, logger)
	if confItem, isConfigurable := item.(Configurable); isConfigurable {
		confItem.Configure(*reader)
	}

	return reader.Errors.OrNil()
}
668424
func (chain *chainImpl) Configure(config *cb.Envelope, configSeq uint64) error {
	return chain.configure(config, configSeq, int64(0))
}
751245
func (writer *influxDBWriter10) configure(conf core.PluginConfigReader, prod *InfluxDB) error {
	writer.host = conf.GetString("Host", "localhost:8086")
	writer.username = conf.GetString("User", "")
	writer.password = conf.GetString("Password", "")
	writer.databaseTemplate = conf.GetString("Database", "default")
	writer.buffer = tio.NewByteStream(4096)
	writer.connectionUp = false
	writer.timeBasedDBName = conf.GetBool("TimeBasedName", true)
	writer.Control = prod.Control
	writer.logger = prod.Logger

	writer.writeURL = fmt.Sprintf("http://%s/write", writer.host)
	writer.queryURL = fmt.Sprintf("http://%s/query", writer.host)
	writer.pingURL = fmt.Sprintf("http://%s/ping", writer.host)
	writer.separator = '?'

	if writer.username != "" {
		credentials := fmt.Sprintf("?u=%s&p=%s", url.QueryEscape(writer.username), url.QueryEscape(writer.password))
		writer.writeURL += credentials
		writer.queryURL += credentials
		writer.separator = '&'
	}

	writer.writeURL = fmt.Sprintf("%s%cprecision=ms", writer.writeURL, writer.separator)
	return conf.Errors.OrNil()
}
338136
func Configure(b *bootstrap.Bootstrapper) {
	h := New(b)
	b.UseGlobal(h)
}
436619
func (format *Identifier) Configure(conf core.PluginConfigReader) {
	idType := strings.ToLower(conf.GetString("Generator", "time"))
	format.seq = new(uint64)

	switch idType {
	case "hash":
		format.hash = format.idHash
	case "seq":
		format.hash = format.idSeq
	case "seqhex":
		format.hash = format.idSeqHex
	case "time":
		format.hash = format.idTime
	default:
		format.hash = func(msg *core.Message) []byte {
			return []byte(idType)
		}
	}
}