bschaeffer

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

bschaeffer

class UpdateCounters < ActiveRecord::Migration
  def up
    say_with_time 'Updating counter_cache_column for ModelWithColumn' do
      ActiveRecord::Base.connection.execute <<-SQL
        UPDATE model_with_column
        SET counter_cache_column = (
          SELECT COUNT(*)
          FROM model_with_counter_cache
          WHERE model_with_counter_cache.reference = model_with_column.id
        )

bschaeffer

class LoadTest
  def initialize(time:, rate:, pool:)
    @time = time
    @rate = rate
    @pool = pool
    @requests = []
    @started = false
    @shutdown = false
    @lock = Mutex.new
    @cond = ConditionVariable.new

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

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,