bschaeffer

apiVersion: install.istio.io/v1alpha1
kind: IstioOperator
metadata:
  name: default
  namespace: istio-system
spec:
  profile: default
  components:
    pilot:
      k8s:

bschaeffer

type logEntry struct {
	Timestamp time.Time `json:"timestamp"`
	Message   string    `json:"message"`
	App       string    `json:"app"`
	Pod       string    `json:"pod"`
}

func buildLogEntry(e publisher.Event) logEntry {
	msg, _ := e.Content.GetValue("message")
	app, _ := e.Content.GetValue("kubernetes.label.app")

bschaeffer

{
  "@timestamp": "2020-06-30T01:23:04.041Z",
  "message": "I am a log line",
  "kubernetes": {
    "pod": {
      "name": "foo-67bd487789"
    },
    "labels": {
      "app": "foo"
    }

bschaeffer

func (h *httpClient) Publish(batch publisher.Batch) error {
	events := batch.Events()
	s.stats.NewBatch(len(events))

	entries := make([]logEntry, 0, len(events))
	for _, event := range events {
		entries = append(entries, buildLogEntry(event))
	}

	var buf bytes.Buffer

bschaeffer

type httpClient struct {
	stats    outputs.Observer
	endpoint string
	client   *http.Client
}

func (h *httpClient) String() string {
	return "fshttp"
}

bschaeffer

type NetworkClient interface {
	String() string
	Connect() error
	Close() error
	Publish(publisher.Batch) error
}

bschaeffer

func newHTTPOutput(_ outputs.IndexManager, _ beat.Info,	stats outputs.Observer, cfg *common.Config) (outputs.Group, error) {
	config := clientConfig{}
	if err := cfg.Unpack(&config); err != nil {
		return outputs.Fail(err)
	}

	clients := make([]outputs.NetworkClient, config.Workers)
	for i := 0; i < config.Workers; i++ {
		clients[i] = &httpClient{
			stats:    stats,

bschaeffer

type clientConfig struct {
	// Number of worker goroutines publishing log events
	Workers int `config:"workers" validate:"min=1"`
	// Max number of events in a batch to send to a single client 
	BatchSize int `config:"batch_size" validate:"min=1"`
	// Max number of retries for single batch of events
	RetryLimit int `config:"retry_limit"`
	// The endpoint our client should be POSTing to
	Endpoint string `config:"endpoint"`
}

bschaeffer

package fshttp

import (
	"github.com/elastic/beats/libbeat/beat"
	"github.com/elastic/beats/libbeat/common"
	"github.com/elastic/beats/libbeat/outputs"
)

func init() {
	outputs.RegisterType("http", newHTTPOutput)

bschaeffer

INFO	instance/beat.go:571	Home path: [/usr/share/filebeat] Config path: [/config] Data path: [/usr/share/filebeat/data] Logs path: [/usr/share/filebeat/logs]
INFO	instance/beat.go:579	Beat ID: ef51dbfd-90fb-4517-bcac-8a4ed45df79c
INFO	[index-management.ilm]	ilm/ilm.go:129	Policy name: filebeat-7.1.1
INFO	add_cloud_metadata/add_cloud_metadata.go:350	add_cloud_metadata: hosting provider type detected as gcp, metadata={"availability_zone":"us-central1-b","instance":{"id":"4300596892232664528","name":"gke-logload-test-custom-bd5d436a-0l3s"},"machine":{"type":"custom-6-15360"},"project":{"id":"fs-ops"},"provider":"gcp"}
INFO	[seccomp]	seccomp/seccomp.go:116	Syscall filter successfully installed
INFO	[beat]	instance/beat.go:827	Beat info	{"system_info": {"beat": {"path": {"config": "/config", "data": "/usr/share/filebeat/data", "home": "/usr/share/filebeat", "logs": "/usr/share/filebeat/logs"}, "type": "filebeat", "uuid": "ef51dbfd-90fb-4517-bcac-8a4ed45df79c"}}}
INFO	[beat]	instance/beat.go:836	Build info	{"system