<commit_before>// Copyright 2015 The LUCI Authors. All rights reserved.
// Use of this source code is governed under the Apache License, Version 2.0
// that can be found in the LICENSE file.

package butler

import (
	"errors"
	"fmt"
	"io"
	"sort"
	"sync"
	"time"

	"github.com/luci/luci-go/client/logdog/butler/bundler"
	"github.com/luci/luci-go/client/logdog/butler/output"
	"github.com/luci/luci-go/client/logdog/butler/streamserver"
	"github.com/luci/luci-go/client/logdog/butlerlib/streamproto"
	"github.com/luci/luci-go/common/clock"
	"github.com/luci/luci-go/common/config"
	"github.com/luci/luci-go/common/iotools"
	"github.com/luci/luci-go/common/logdog/types"
	log "github.com/luci/luci-go/common/logging"
	"github.com/luci/luci-go/common/paniccatcher"
	"github.com/luci/luci-go/common/parallel"
	"github.com/luci/luci-go/common/proto/google"
	"github.com/luci/luci-go/common/stringset"
	"golang.org/x/net/context"
)

const (
	// DefaultMaxBufferAge is the default amount of time that a log entry may
	// be buffered before being dispatched.
	DefaultMaxBufferAge = time.Duration(5 * time.Second)

	// DefaultOutputWorkers is the default number of output workers to use.
	DefaultOutputWorkers = 16

	// streamBufferSize is the maximum amount of stream data to buffer in memory.
	streamBufferSize = 1024 * 1024 * 5
)

// Config is the set of Butler configuration parameters.
type Config struct {
	// Output is the output instance to use for log dispatch.
	Output output.Output
	// OutputWorkers is the number of simultaneous goroutines that will be used
	// to output Butler log data. If zero, DefaultOutputWorkers will be used.
	OutputWorkers int

	// Project is the project that the log stream will be bound to.
	Project config.ProjectName
	// Prefix is the log stream common prefix value.
	Prefix types.StreamName

	// BufferLogs, if true, instructs the butler to buffer collected log data
	// before sending it to Output.
	BufferLogs bool
	// If buffering logs, this is the maximum amount of time that a log will
	// be buffered before being marked for dispatch. If this is zero,
	// DefaultMaxBufferAge will be used.
	MaxBufferAge time.Duration

	// TeeStdout, if not nil, is the Writer that will be used for streams
	// requesting STDOUT tee.
	TeeStdout io.Writer
	// TeeStderr, if not nil, is the Writer that will be used for streams
	// requesting STDERR tee.
	TeeStderr io.Writer
}

// Validate validates that the configuration is sufficient to instantiate a
// Butler instance.
func (c *Config) Validate() error {
	if c.Output == nil {
		return errors.New("butler: an Output must be supplied")
	}
	if err := c.Project.Validate(); err != nil {
		return fmt.Errorf("invalid project: %v", err)
	}
	if err := c.Prefix.Validate(); err != nil {
		return fmt.Errorf("invalid prefix: %v", err)
	}
	return nil
}

// Butler is the Butler application structure. The Butler runs until closed.
// During operation, it acts as a service manager and data router, routing:
// - Messages from Streams to the attached Output.
// - Streams from a StreamServer to the Stream list (AddStream).
type Butler struct {
	c   *Config
	ctx context.Context

	// bundler is the Bundler instance.
	bundler *bundler.Bundler
	// bundlerDrainedC is a signal channel that is closed when the Bundler has
	// been drained.
	bundlerDrainedC chan struct{}

	// activateC is closed when Activate() is called.
	activateC chan struct{}
	// activateOnce ensures we close activeC exactly once.
	activateOnce sync.Once

	// streamsFinishedC is a signal channel that will be closed when the stream
	// monitor finishes its managed stream set.
	streamsFinishedC chan struct{}

	// WaitGroup to manage running StreamServers.
	streamServerWG sync.WaitGroup
	// Channel to signal StreamServers to stop.
	streamServerStopC chan struct{}

	streamC chan *stream
	// streamSeen tracks if a stream has been seen. This is used to prevent
	// duplicate stream names from being created.
	streamSeen stringset.Set
	// streamSeenLock is a lock to protect streamSeen.
	streamSeenLock sync.Mutex

	// shutdownMu is a mutex to protect shutdown parameters.
	shutdownMu sync.Mutex
	// isShutdown is true if the Butler been shut down.
	isShutdown bool
	// runErr is the error returned by Run.
	runErr error
	// streamStopC is a stop signal channel for stream. This will cause streams
	// to prematurely terminate (before EOF) on shutdown.
	streamStopC chan struct{}
}

// New instantiates a new Butler instance and starts its processing.
func New(ctx context.Context, config Config) (*Butler, error) {
	if err := config.Validate(); err != nil {
		return nil, err
	}

	if config.OutputWorkers <= 0 {
		config.OutputWorkers = DefaultOutputWorkers
	}

	bc := bundler.Config{
		Clock:            clock.Get(ctx),
		Project:          config.Project,
		Prefix:           config.Prefix,
		MaxBufferedBytes: streamBufferSize,
		MaxBundleSize:    config.Output.MaxSize(),
	}
	if config.BufferLogs {
		bc.MaxBufferDelay = config.MaxBufferAge
	}<commit_msg>LogDog: Actually use default max buffer age.<commit_after>// Copyright 2015 The LUCI Authors. All rights reserved.
// Use of this source code is governed under the Apache License, Version 2.0
// that can be found in the LICENSE file.

package butler

import (
	"errors"
	"fmt"
	"io"
	"sort"
	"sync"
	"time"

	"github.com/luci/luci-go/client/logdog/butler/bundler"
	"github.com/luci/luci-go/client/logdog/butler/output"
	"github.com/luci/luci-go/client/logdog/butler/streamserver"
	"github.com/luci/luci-go/client/logdog/butlerlib/streamproto"
	"github.com/luci/luci-go/common/clock"
	"github.com/luci/luci-go/common/config"
	"github.com/luci/luci-go/common/iotools"
	"github.com/luci/luci-go/common/logdog/types"
	log "github.com/luci/luci-go/common/logging"
	"github.com/luci/luci-go/common/paniccatcher"
	"github.com/luci/luci-go/common/parallel"
	"github.com/luci/luci-go/common/proto/google"
	"github.com/luci/luci-go/common/stringset"
	"golang.org/x/net/context"
)

const (
	// DefaultMaxBufferAge is the default amount of time that a log entry may
	// be buffered before being dispatched.
	DefaultMaxBufferAge = time.Duration(5 * time.Second)

	// DefaultOutputWorkers is the default number of output workers to use.
	DefaultOutputWorkers = 16

	// streamBufferSize is the maximum amount of stream data to buffer in memory.
	streamBufferSize = 1024 * 1024 * 5
)

// Config is the set of Butler configuration parameters.
type Config struct {
	// Output is the output instance to use for log dispatch.
	Output output.Output
	// OutputWorkers is the number of simultaneous goroutines that will be used
	// to output Butler log data. If zero, DefaultOutputWorkers will be used.
	OutputWorkers int

	// Project is the project that the log stream will be bound to.
	Project config.ProjectName
	// Prefix is the log stream common prefix value.
	Prefix types.StreamName

	// BufferLogs, if true, instructs the butler to buffer collected log data
	// before sending it to Output.
	BufferLogs bool
	// If buffering logs, this is the maximum amount of time that a log will
	// be buffered before being marked for dispatch. If this is zero,
	// DefaultMaxBufferAge will be used.
	MaxBufferAge time.Duration

	// TeeStdout, if not nil, is the Writer that will be used for streams
	// requesting STDOUT tee.
	TeeStdout io.Writer
	// TeeStderr, if not nil, is the Writer that will be used for streams
	// requesting STDERR tee.
	TeeStderr io.Writer
}

// Validate validates that the configuration is sufficient to instantiate a
// Butler instance.
func (c *Config) Validate() error {
	if c.Output == nil {
		return errors.New("butler: an Output must be supplied")
	}
	if err := c.Project.Validate(); err != nil {
		return fmt.Errorf("invalid project: %v", err)
	}
	if err := c.Prefix.Validate(); err != nil {
		return fmt.Errorf("invalid prefix: %v", err)
	}
	return nil
}

// Butler is the Butler application structure. The Butler runs until closed.
// During operation, it acts as a service manager and data router, routing:
// - Messages from Streams to the attached Output.
// - Streams from a StreamServer to the Stream list (AddStream).
type Butler struct {
	c   *Config
	ctx context.Context

	// bundler is the Bundler instance.
	bundler *bundler.Bundler
	// bundlerDrainedC is a signal channel that is closed when the Bundler has
	// been drained.
	bundlerDrainedC chan struct{}

	// activateC is closed when Activate() is called.
	activateC chan struct{}
	// activateOnce ensures we close activeC exactly once.
	activateOnce sync.Once

	// streamsFinishedC is a signal channel that will be closed when the stream
	// monitor finishes its managed stream set.
	streamsFinishedC chan struct{}

	// WaitGroup to manage running StreamServers.
	streamServerWG sync.WaitGroup
	// Channel to signal StreamServers to stop.
	streamServerStopC chan struct{}

	streamC chan *stream
	// streamSeen tracks if a stream has been seen. This is used to prevent
	// duplicate stream names from being created.
	streamSeen stringset.Set
	// streamSeenLock is a lock to protect streamSeen.
	streamSeenLock sync.Mutex

	// shutdownMu is a mutex to protect shutdown parameters.
	shutdownMu sync.Mutex
	// isShutdown is true if the Butler been shut down.
	isShutdown bool
	// runErr is the error returned by Run.
	runErr error
	// streamStopC is a stop signal channel for stream. This will cause streams
	// to prematurely terminate (before EOF) on shutdown.
	streamStopC chan struct{}
}

// New instantiates a new Butler instance and starts its processing.
func New(ctx context.Context, config Config) (*Butler, error) {
	if err := config.Validate(); err != nil {
		return nil, err
	}

	if config.OutputWorkers <= 0 {
		config.OutputWorkers = DefaultOutputWorkers
	}

	bc := bundler.Config{
		Clock:            clock.Get(ctx),
		Project:          config.Project,
		Prefix:           config.Prefix,
		MaxBufferedBytes: streamBufferSize,
		MaxBundleSize:    config.Output.MaxSize(),
	}
	if config.BufferLogs {
		bc.MaxBufferDelay = config.MaxBufferAge
		if bc.MaxBufferDelay <= 0 {
			bc.MaxBufferDelay = DefaultMaxBufferAge
		}
	}
{'fraction_non_alphanumeric': 0.08879429815878044, 'fraction_numerical': 0.0038606216590774105, 'mean_word_length': 3.5881017257039054, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6934728', 'n_tokens_mistral': 3238, 'n_tokens_neox': 3093, 'n_words': 1141}