Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Journalbeat: add index option to input #15071

Merged
merged 11 commits into from
Dec 13, 2019
30 changes: 2 additions & 28 deletions filebeat/channel/connector.go
Original file line number Diff line number Diff line change
@@ -18,12 +18,11 @@
package channel

import (
"fmt"

"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/fmtstr"
"github.com/elastic/beats/libbeat/processors"
"github.com/elastic/beats/libbeat/processors/add_formatted_index"
)

// ConnectorFunc is an adapter for using ordinary functions as Connector.
@@ -34,14 +33,6 @@ type pipelineConnector struct {
pipeline beat.Pipeline
}

// addFormattedIndex is a Processor to set an event's "raw_index" metadata field
// with a given TimestampFormatString. The elasticsearch output interprets
// that field as specifying the (raw string) index the event should be sent to;
// in other outputs it is just included in the metadata.
type addFormattedIndex struct {
formatString *fmtstr.TimestampFormatString
}

// Connect passes the cfg and the zero value of beat.ClientConfig to the underlying function.
func (fn ConnectorFunc) Connect(cfg *common.Config) (Outleter, error) {
return fn(cfg, beat.ClientConfig{})
@@ -132,7 +123,7 @@ func processorsForConfig(
if err != nil {
return nil, err
}
indexProcessor := &addFormattedIndex{timestampFormat}
indexProcessor := add_formatted_index.New(timestampFormat)
procs.List = append(procs.List, indexProcessor)
}

@@ -160,20 +151,3 @@ func processorsForConfig(
procs.List = append(procs.List, userProcessors.List...)
return procs, nil
}

func (p *addFormattedIndex) Run(event *beat.Event) (*beat.Event, error) {
index, err := p.formatString.Run(event.Timestamp)
if err != nil {
return nil, err
}

if event.Meta == nil {
event.Meta = common.MapStr{}
}
event.Meta["raw_index"] = index
return event, nil
}

func (p *addFormattedIndex) String() string {
return fmt.Sprintf("add_index_pattern=%v", p.formatString)
}
2 changes: 1 addition & 1 deletion journalbeat/beater/journalbeat.go
Original file line number Diff line number Diff line change
@@ -67,7 +67,7 @@ func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {

var inputs []*input.Input
for _, c := range config.Inputs {
i, err := input.New(c, b.Publisher, done, cp.States())
i, err := input.New(c, b, done, cp.States())
if err != nil {
return nil, err
}
3 changes: 3 additions & 0 deletions journalbeat/input/config.go
Original file line number Diff line number Diff line change
@@ -22,6 +22,7 @@ import (

"github.com/elastic/beats/journalbeat/config"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/fmtstr"
"github.com/elastic/beats/libbeat/processors"
)

@@ -47,6 +48,8 @@ type Config struct {
common.EventMetadata `config:",inline"`
// Processors to run on events.
Processors processors.PluginConfig `config:"processors"`
// ES output index pattern
Index fmtstr.EventFormatString `config:"index"`
}

var (
49 changes: 45 additions & 4 deletions journalbeat/input/input.go
Original file line number Diff line number Diff line change
@@ -21,6 +21,10 @@ import (
"fmt"
"sync"

"github.com/elastic/beats/libbeat/processors/add_formatted_index"

"github.com/elastic/beats/libbeat/common/fmtstr"

"github.com/gofrs/uuid"

"github.com/elastic/beats/journalbeat/checkpoint"
@@ -48,7 +52,7 @@ type Input struct {
// New returns a new Inout
func New(
c *common.Config,
pipeline beat.Pipeline,
b *beat.Beat,
done chan struct{},
states map[string]checkpoint.JournalState,
) (*Input, error) {
@@ -102,7 +106,7 @@ func New(
readers = append(readers, r)
}

processors, err := processors.New(config.Processors)
inputProcessors, err := processorsForInput(b.Info, config)
if err != nil {
return nil, err
}
@@ -113,12 +117,12 @@ func New(
readers: readers,
done: done,
config: config,
pipeline: pipeline,
pipeline: b.Publisher,
states: states,
id: id,
logger: logger,
eventMeta: config.EventMetadata,
processors: processors,
processors: inputProcessors,
}, nil
}

@@ -203,3 +207,40 @@ func (i *Input) Stop() {
func (i *Input) Wait() {
i.Stop()
}

func processorsForInput(beatInfo beat.Info, config Config) (*processors.Processors, error) {
procs := processors.NewList(nil)

// Processor ordering is important:
// 1. Index configuration
if !config.Index.IsEmpty() {
staticFields := fmtstr.FieldsForBeat(beatInfo.Beat, beatInfo.Version)
timestampFormat, err :=
fmtstr.NewTimestampFormatString(&config.Index, staticFields)
if err != nil {
return nil, err
}
indexProcessor := add_formatted_index.New(timestampFormat)
procs.List = append(procs.List, indexProcessor)
}

// 2. User processors
userProcessors, err := processors.New(config.Processors)
if err != nil {
return nil, err
}
// Subtlety: it is important here that we append the individual elements of
// userProcessors, rather than userProcessors itself, even though
// userProcessors implements the processors.Processor interface. This is
// because the contents of what we return are later pulled out into a
// processing.group rather than a processors.Processors, and the two have
// different error semantics: processors.Processors aborts processing on
// any error, whereas processing.group only aborts on fatal errors. The
// latter is the most common behavior, and the one we are preserving here for
// backwards compatibility.
// We are unhappy about this and have plans to fix this inconsistency at a
// higher level, but for now we need to respect the existing semantics.
procs.List = append(procs.List, userProcessors.List...)

return procs, nil
}
55 changes: 55 additions & 0 deletions libbeat/processors/add_formatted_index/add_formatted_index.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

package add_formatted_index
ycombinator marked this conversation as resolved.
Show resolved Hide resolved

import (
"fmt"

"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/fmtstr"
)

// addFormattedIndex is a Processor to set an event's "raw_index" metadata field
// with a given TimestampFormatString. The elasticsearch output interprets
// that field as specifying the (raw string) index the event should be sent to;
// in other outputs it is just included in the metadata.
type addFormattedIndex struct {
formatString *fmtstr.TimestampFormatString
}

func New(formatString *fmtstr.TimestampFormatString) *addFormattedIndex {
ycombinator marked this conversation as resolved.
Show resolved Hide resolved
return &addFormattedIndex{formatString}
}

func (p *addFormattedIndex) Run(event *beat.Event) (*beat.Event, error) {
index, err := p.formatString.Run(event.Timestamp)
if err != nil {
return nil, err
}

if event.Meta == nil {
event.Meta = common.MapStr{}
}
event.Meta["raw_index"] = index
return event, nil
}

func (p *addFormattedIndex) String() string {
return fmt.Sprintf("add_index_pattern=%v", p.formatString)
}