// NewSyslogLineParser factory
331339
func NewSyslogLineParser(mode string, processList string) (*SyslogLineParser, error) {
	var supportedProcesses map[string]struct{}
	// if a list of process
	if processList != "" {
		supportedProcesses = make(map[string]struct{})
		for _, process := range strings.Split(processList, ",") {
			supportedProcesses[strings.TrimSpace(process)] = struct{}{}
		}
	}
	if mode == "rfc3164" || mode == "rfc5424" {
		return &SyslogLineParser{
			mode:               mode,
			supportedProcesses: supportedProcesses,
		}, nil
	}

	return nil, fmt.Errorf("unsupported mode %s, see --help", mode)
}
511331
func parseLogSource(date string, source []byte) (*parsedMessage, error) {
	warnings := ""

	// attempt to unmarshal into singleLine
	var line logSingleLine

	if e := json.Unmarshal(source, &line); e == nil {
		offset := uint64(0)
		if len(line.Offset) != 0 {
			var e error
			offset, e = strconv.ParseUint(string(line.Offset), 10, 64)
			if e != nil {
				return nil, fmt.Errorf("failed to parse offset \"%s\" in \"%s\": %s", line.Offset, source, e)
			}
		}
		compactLine := compactLogLine{
			Timestamp: truncateToMinute(line.Timestamp.UnixNano()),
			Offset:    offset,
			Message:   line.Message,
		}

		// Get the ccWorkerID. This can be a number in older systems, so we have to parse it out.
		ccWorkerID, err := convertWorkerID(line.Fields.CCWorkerID)
		if err != nil {
			return nil, fmt.Errorf("failed to parse ccWorkerID from %v in \"%s\": %s", line.Fields.CCWorkerID, source, err)
		}

		message := &parsedMessage{
			Date:        date,
			Type:        line.Type,
			HostID:      ccWorkerID,
			ContainerID: line.FileBeat.Hostname,
			LogFileName: line.File,
			Lines:       []compactLogLine{compactLine},
			ServiceID:   line.Fields.Service,
		}
		return message, nil
	}

	// attempt to unmarshal into multiLine
	multiLine, err := convertMultiLineSource(source)
	if err != nil {
		return nil, fmt.Errorf("failed to parse multiLine log from JSON \"%s\": %s", source, err)
	}

	// build offsets - list of uint64
	offsets := multiLine.Offsets

	// verify number of lines in message against number of offsets
	messages := multiLine.Messages
	if len(offsets)+1 == len(messages) {
		warnings += fmt.Sprintf(
			"number of offsets for %s:%s (numLines:%d numOffsets:%d) is one less than number of lines: %s\n",
			multiLine.FileBeat.Hostname, multiLine.File, len(messages), len(offsets), source)
		numLines := len(messages)
		if numLines > 1 {
			lastOffset := uint64(len(messages[numLines-2])) + offsets[numLines-2]
			offsets = append(offsets, lastOffset)
		}
	} else if len(offsets) > len(messages) {
		warnings += fmt.Sprintf(
			"number of offsets for %s:%s (numLines:%d numOffsets:%d) is greater than number of lines: %s\n",
			multiLine.FileBeat.Hostname, multiLine.File, len(messages), len(multiLine.Offsets), source)
		offsets = offsets[0:len(messages)]
	} else if len(offsets) < len(messages) {
		warnings += fmt.Sprintf(
			"number of offsets for %s:%s (numLines:%d numOffsets:%d) is less than number of lines: %s\n",
			multiLine.FileBeat.Hostname, multiLine.File, len(messages), len(multiLine.Offsets), source)
		offsets = generateOffsets(messages, offsets)
		warnings += fmt.Sprintf("new offsets: %v\n", offsets)
	}

	// deal with offsets that are not sorted in increasing order
	if !uint64sAreSorted(offsets) {
		warnings += fmt.Sprintf("offsets are not sorted: %v\n", offsets)
		offsets = generateOffsets(messages, offsets)
		warnings += fmt.Sprintf("new offsets: %v\n", offsets)
	}

	// build compactLines
	timestamp := truncateToMinute(multiLine.Timestamp.UnixNano())
	compactLines := make([]compactLogLine, len(messages))
	for i, offset := range offsets {
		compactLines = append(compactLines, compactLogLine{
			Timestamp: timestamp,
			Offset:    offset,
			Message:   messages[i],
		})
	}

	// Get the ccWorkerID. This can be a number in older systems, so we have to parse it out.
	ccWorkerID, err := convertWorkerID(multiLine.Fields.CCWorkerID)
	if err != nil {
		return nil, fmt.Errorf("failed to parse ccWorkerID from %v in \"%s\": %s", multiLine.Fields.CCWorkerID, source, err)
	}

	message := &parsedMessage{
		Date:        date,
		Type:        line.Type,
		HostID:      ccWorkerID,
		ContainerID: multiLine.FileBeat.Hostname,
		LogFileName: multiLine.File,
		Lines:       compactLines,
		Warnings:    warnings,
		ServiceID:   multiLine.Fields.Service,
	}
	return message, nil
}
661866
func Syslogf(p Priority, format string, a ...interface{}) {
	Syslog(p, fmt.Sprintf(format, a...))
}
52425
func NewLogFmtParser(metricName string, defaultTags map[string]string) (Parser, error) {
	return logfmt.NewParser(metricName, defaultTags), nil
}
1105047
func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event, prefixRegex *parsers.ExtRegexp) {
	wg := sync.WaitGroup{}
	for i := 0; i < numParsers; i++ {
		wg.Add(1)
		go func() {
			for line := range lines {
				line = strings.TrimSpace(line)
				// take care of any headers on the line
				var prefixFields map[string]string
				if prefixRegex != nil {
					var prefix string
					prefix, prefixFields = prefixRegex.FindStringSubmatchMap(line)
					line = strings.TrimPrefix(line, prefix)
				}

				values, err := p.lineParser.ParseLine(line)
				// we get a bunch of errors from the parser on ArangoDB logs, skip em
				if err == nil {
					timestamp, err := p.parseTimestamp(values)
					if err != nil {
						logSkipped(line, "couldn't parse logline timestamp, skipping")
						continue
					}

					// merge the prefix fields and the parsed line contents
					for k, v := range prefixFields {
						values[k] = v
					}

					logrus.WithFields(logrus.Fields{
						"line":   line,
						"values": values,
					}).Debug("Successfully parsed line")

					// we'll be putting the timestamp in the Event
					// itself, no need to also have it in the Data
					delete(values, timestampFieldName)

					send <- event.Event{
						Timestamp: timestamp,
						Data:      values,
					}
				} else {
					logSkipped(line, "logline didn't parse, skipping.")
				}
			}
			wg.Done()
		}()
	}
	wg.Wait()
	logrus.Debug("lines channel is closed, ending arangodb processor")
}
244686
func NewStoragePacker(view logical.Storage, logger log.Logger, viewPrefix string) (*StoragePacker, error) {
	if view == nil {
		return nil, fmt.Errorf("nil view")
	}

	if viewPrefix == "" {
		viewPrefix = StoragePackerBucketsPrefix
	}

	if !strings.HasSuffix(viewPrefix, "/") {
		viewPrefix = viewPrefix + "/"
	}

	// Create a new packer object for the given view
	packer := &StoragePacker{
		view:         view,
		viewPrefix:   viewPrefix,
		logger:       logger,
		storageLocks: locksutil.CreateLocks(),
	}

	return packer, nil
}
1196484
func (cl *ChannelLogger) FlushLogLines() [][]interface{} {
	logLines := make([][]interface{}, 0, cl.ch.Len())
	cl.Flush(log.LoggerFunc(func(keyvals ...interface{}) error {
		logLines = append(logLines, keyvals)
		return nil
	}))
	return logLines
}
730104
func NewSyslogOutput(facility SyslogPriority, tag string) (
	TextOutput, error) {
	w, err := syslog.New(syslog.Priority(facility), tag)
	if err != nil {
		return nil, err
	}
	return &SyslogOutput{w: w}, nil
}
1335919
func NewRegexLineParser(regexStrs []string) (*RegexLineParser, error) {
	lineRegexes, err := ParseLineRegexes(regexStrs)
	if err != nil {
		return nil, err
	}
	logrus.WithFields(logrus.Fields{
		"lineRegexes": lineRegexes,
	}).Debug("Compiled line regexes")
	return &RegexLineParser{lineRegexes}, nil
}
578022
func (c *ReportingClient) RunStepLogs(ctx context.Context, args RunStepLogsArgs) error {
	return c.postJSON("runsteplogs", args)
}
1293784
func (fw *PaloAlto) sendSyslog(ctx context.Context, line string) error {
	writer, err := fw.getSyslog(ctx)

	if err != nil {
		return err
	}

	err = writer.Err(line)

	if err != nil {
		log.LoggerWContext(ctx).Error(fmt.Sprintf("Error sending message to PaloAlto: %s", err))
		return err
	}

	return nil
}
934681
func OpenSyslog(cfg *syslog.RawConfig) (*logforwarder.LogSink, error) {
	if !cfg.Enabled {
		return nil, errors.New("log forwarding not enabled")
	}
	client, err := syslog.Open(*cfg)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if client == nil {
		// TODO(axw) we should be returning an error
		// which we interpret up the stack.
		return &logforwarder.LogSink{
			SendCloser: emptySendCloser{},
		}, nil
	}
	sink := &logforwarder.LogSink{
		SendCloser: client,
	}
	return sink, nil
}
156786
func NewServiceWithLogger(serviceName string, client *api.Client,
	logger *log.Logger) (*Service, error) {
	s := &Service{
		service:              serviceName,
		client:               client,
		logger:               logger,
		tlsCfg:               newDynamicTLSConfig(defaultTLSConfig(), logger),
		httpResolverFromAddr: ConsulResolverFromAddrFunc(client),
	}

	// Set up root and leaf watches
	p, err := watch.Parse(map[string]interface{}{
		"type": "connect_roots",
	})
	if err != nil {
		return nil, err
	}
	s.rootsWatch = p
	s.rootsWatch.HybridHandler = s.rootsWatchHandler

	p, err = watch.Parse(map[string]interface{}{
		"type":    "connect_leaf",
		"service": s.service,
	})
	if err != nil {
		return nil, err
	}
	s.leafWatch = p
	s.leafWatch.HybridHandler = s.leafWatchHandler

	go s.rootsWatch.RunWithClientAndLogger(client, s.logger)
	go s.leafWatch.RunWithClientAndLogger(client, s.logger)

	return s, nil
}
526329
func NewSysLogger(tag string) (Logger, error) {
	writer, err := syslog.New(syslog.LOG_DEBUG|syslog.LOG_LOCAL0, tag)
	if err != nil {
		log.Fatalf("cannot init syslog: %v", err)
		return nil, err
	}
	return &SysLogger{writer}, nil
}
696402
func MultiLine(logFn func(args ...interface{}), output string) {
	scanner := bufio.NewScanner(bytes.NewReader([]byte(output)))
	for scanner.Scan() {
		logFn(scanner.Text())
	}
}
952930
func NewStringLiteral(pos int, line int, val string) *StringLiteral {
	return &StringLiteral{
		NodeType: NodeString,
		Loc:      Loc{pos, line},

		Value: val,
	}
}
393418
func (p *SlowLogParser) parseHeader(line string) {
	if p.opt.Debug {
		l.Println("header")
	}

	if !headerRe.MatchString(line) {
		p.inHeader = false
		p.inQuery = true
		p.parseQuery(line)
		return
	}

	if p.headerLines == 0 {
		p.event.Offset = p.lineOffset
	}
	p.headerLines++

	if strings.HasPrefix(line, "# Time") {
		if p.opt.Debug {
			l.Println("time")
		}
		m := timeRe.FindStringSubmatch(line)
		if len(m) == 2 {
			p.event.Ts, _ = time.ParseInLocation("060102 15:04:05", m[1], p.opt.DefaultLocation)
		} else {
			m = timeNewRe.FindStringSubmatch(line)
			if len(m) == 2 {
				p.event.Ts, _ = time.ParseInLocation(time.RFC3339Nano, m[1], p.opt.DefaultLocation)
			} else {
				return
			}
		}
		if userRe.MatchString(line) {
			if p.opt.Debug {
				l.Println("user (bad format)")
			}
			m := userRe.FindStringSubmatch(line)
			p.event.User = m[1]
			p.event.Host = m[2]
		}
	} else if strings.HasPrefix(line, "# User") {
		if p.opt.Debug {
			l.Println("user")
		}
		m := userRe.FindStringSubmatch(line)
		if len(m) < 3 {
			return
		}
		p.event.User = m[1]
		p.event.Host = m[2]
	} else if strings.HasPrefix(line, "# admin") {
		p.parseAdmin(line)
	} else {
		if p.opt.Debug {
			l.Println("metrics")
		}
		submatch := schema.FindStringSubmatch(line)
		if len(submatch) == 2 {
			p.event.Db = submatch[1]
		}

		m := metricsRe.FindAllStringSubmatch(line, -1)
		for _, smv := range m {
			// [String, Metric, Value], e.g. ["Query_time: 2", "Query_time", "2"]
			if strings.HasSuffix(smv[1], "_time") || strings.HasSuffix(smv[1], "_wait") {
				// microsecond value
				val, _ := strconv.ParseFloat(smv[2], 64)
				p.event.TimeMetrics[smv[1]] = val
			} else if smv[2] == "Yes" || smv[2] == "No" {
				// boolean value
				if smv[2] == "Yes" {
					p.event.BoolMetrics[smv[1]] = true
				} else {
					p.event.BoolMetrics[smv[1]] = false
				}
			} else if smv[1] == "Schema" {
				p.event.Db = smv[2]
			} else if smv[1] == "Log_slow_rate_type" {
				p.event.RateType = smv[2]
			} else if smv[1] == "Log_slow_rate_limit" {
				val, _ := strconv.ParseUint(smv[2], 10, 64)
				p.event.RateLimit = uint(val)
			} else {
				// integer value
				val, _ := strconv.ParseUint(smv[2], 10, 64)
				p.event.NumberMetrics[smv[1]] = val
			}
		}
	}
}