<commit_before>package burrow

import (
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/config"
	"github.com/influxdata/telegraf/filter"
	"github.com/influxdata/telegraf/plugins/common/tls"
	"github.com/influxdata/telegraf/plugins/inputs"
)

const (
	defaultBurrowPrefix          = "/v3/kafka"
	defaultConcurrentConnections = 20
	defaultResponseTimeout       = time.Second * 5
	defaultServer                = "http://localhost:8000"
)

type (
	burrow struct {
		tls.ClientConfig

		Servers               []string
		Username              string
		Password              string
		ResponseTimeout       config.Duration
		ConcurrentConnections int

		APIPrefix       string `toml:"api_prefix"`
		ClustersExclude []string
		ClustersInclude []string
		GroupsExclude   []string
		GroupsInclude   []string
		TopicsExclude   []string
		TopicsInclude   []string

		client         *http.Client
		filterClusters filter.Filter
		filterGroups   filter.Filter
		filterTopics   filter.Filter
	}

	// response
	apiResponse struct {
		Clusters []string          `json:"clusters"`
		Groups   []string          `json:"consumers"`
		Topics   []string          `json:"topics"`
		Offsets  []int64           `json:"offsets"`
		Status   apiStatusResponse `json:"status"`
	}

	// response: status field
	apiStatusResponse struct {
		Partitions     []apiStatusResponseLag `json:"partitions"`
		Status         string                 `json:"status"`
		PartitionCount int                    `json:"partition_count"`
		Maxlag         *apiStatusResponseLag  `json:"maxlag"`
		TotalLag       int64                  `json:"totallag"`
	}

	// response: lag field
	apiStatusResponseLag struct {
		Topic      string                   `json:"topic"`
		Partition  int32                    `json:"partition"`
		Status     string                   `json:"status"`
		Start      apiStatusResponseLagItem `json:"start"`
		End        apiStatusResponseLagItem `json:"end"`
		CurrentLag int64                    `json:"current_lag"`
		Owner      string                   `json:"owner"`
	}

	// response: lag field item
	apiStatusResponseLagItem struct {
		Offset    int64 `json:"offset"`
		Timestamp int64 `json:"timestamp"`
		Lag       int64 `json:"lag"`
	}
)

func init() {
	inputs.Add("burrow", func() telegraf.Input {
		return &burrow{}
	})
}

func (b *burrow) Gather(acc telegraf.Accumulator) error {
	var wg sync.WaitGroup

	if len(b.Servers) == 0 {
		b.Servers = []string{defaultServer}
	}

	if b.client == nil {
		b.setDefaults()
		if err := b.compileGlobs(); err != nil {
			return err
		}
		c, err := b.createClient()
		if err != nil {
			return err
		}
		b.client = c
	}

	for _, addr := range b.Servers {
		u, err := url.Parse(addr)
		if err != nil {
			acc.AddError(fmt.Errorf("unable to parse address '%s': %s", addr, err))
			continue
		}
		if u.Path == "" {
			u.Path = b.APIPrefix
		}

		wg.Add(1)
		go func(u *url.URL) {
			defer wg.Done()
			acc.AddError(b.gatherServer(u, acc))
		}(u)
	}

	wg.Wait()
	return nil
}

func (b *burrow) setDefaults() {
	if b.APIPrefix == "" {
		b.APIPrefix = defaultBurrowPrefix
	}
	if b.ConcurrentConnections < 1 {
		b.ConcurrentConnections = defaultConcurrentConnections
	}
	if time.Duration(b.ResponseTimeout) < time.Second {
		b.ResponseTimeout = config.Duration(defaultResponseTimeout)
	}
}

func (b *burrow) compileGlobs() error {
	var err error

	// compile glob patterns
	b.filterClusters, err = filter.NewIncludeExcludeFilter(b.ClustersInclude, b.ClustersExclude)
	if err != nil {
		return err
	}
	b.filterGroups, err = filter.NewIncludeExcludeFilter(b.GroupsInclude, b.GroupsExclude)
	if err != nil {
		return err
	}
	b.filterTopics, err = filter.NewIncludeExcludeFilter(b.TopicsInclude, b.TopicsExclude)
	if err != nil {
		return err
	}
	return nil
}

func (b *burrow) createClient() (*http.Client, error) {
	tlsCfg, err := b.ClientConfig.TLSConfig()
	if err != nil {
		return nil, err
	}

	timeout := time.Duration(b.ResponseTimeout)
	transport := http.Transport{
		DialContext: &net.Dialer{Timeout: timeout, DualStack: true}.DialContext,
		TLSClientConfig: tlsCfg,
		// If b.ConcurrentConnections <= 1, then DefaultMaxIdleConnsPerHost is used (=2)
		MaxIdleConnsPerHost: b.ConcurrentConnections / 2,
		// If b.ConcurrentConnections == 0, then it is treated as "no limits"
		MaxConnsPerHost: b.ConcurrentConnections,
		ResponseHeaderTimeout: timeout,
		IdleConnTimeout: 90*time.Second,
	}

	client := &http.Client{
		Transport: &transport,
		Timeout: timeout,
	}

	return client, nil
}

func (b *burrow) getResponse(u *url.URL) (*apiResponse, error) {
	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
	if err != nil {
		return nil, err
	}
	if b.Username != "" {
		req.SetBasicAuth(b.Username, b.Password)<commit_msg>fix(inputs.burrow): Move Dialer to variable and run `make fmt` (#11149)<commit_after>package burrow

import (
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/config"
	"github.com/influxdata/telegraf/filter"
	"github.com/influxdata/telegraf/plugins/common/tls"
	"github.com/influxdata/telegraf/plugins/inputs"
)

const (
	defaultBurrowPrefix          = "/v3/kafka"
	defaultConcurrentConnections = 20
	defaultResponseTimeout       = time.Second * 5
	defaultServer                = "http://localhost:8000"
)

type (
	burrow struct {
		tls.ClientConfig

		Servers               []string
		Username              string
		Password              string
		ResponseTimeout       config.Duration
		ConcurrentConnections int

		APIPrefix       string `toml:"api_prefix"`
		ClustersExclude []string
		ClustersInclude []string
		GroupsExclude   []string
		GroupsInclude   []string
		TopicsExclude   []string
		TopicsInclude   []string

		client         *http.Client
		filterClusters filter.Filter
		filterGroups   filter.Filter
		filterTopics   filter.Filter
	}

	// response
	apiResponse struct {
		Clusters []string          `json:"clusters"`
		Groups   []string          `json:"consumers"`
		Topics   []string          `json:"topics"`
		Offsets  []int64           `json:"offsets"`
		Status   apiStatusResponse `json:"status"`
	}

	// response: status field
	apiStatusResponse struct {
		Partitions     []apiStatusResponseLag `json:"partitions"`
		Status         string                 `json:"status"`
		PartitionCount int                    `json:"partition_count"`
		Maxlag         *apiStatusResponseLag  `json:"maxlag"`
		TotalLag       int64                  `json:"totallag"`
	}

	// response: lag field
	apiStatusResponseLag struct {
		Topic      string                   `json:"topic"`
		Partition  int32                    `json:"partition"`
		Status     string                   `json:"status"`
		Start      apiStatusResponseLagItem `json:"start"`
		End        apiStatusResponseLagItem `json:"end"`
		CurrentLag int64                    `json:"current_lag"`
		Owner      string                   `json:"owner"`
	}

	// response: lag field item
	apiStatusResponseLagItem struct {
		Offset    int64 `json:"offset"`
		Timestamp int64 `json:"timestamp"`
		Lag       int64 `json:"lag"`
	}
)

func init() {
	inputs.Add("burrow", func() telegraf.Input {
		return &burrow{}
	})
}

func (b *burrow) Gather(acc telegraf.Accumulator) error {
	var wg sync.WaitGroup

	if len(b.Servers) == 0 {
		b.Servers = []string{defaultServer}
	}

	if b.client == nil {
		b.setDefaults()
		if err := b.compileGlobs(); err != nil {
			return err
		}
		c, err := b.createClient()
		if err != nil {
			return err
		}
		b.client = c
	}

	for _, addr := range b.Servers {
		u, err := url.Parse(addr)
		if err != nil {
			acc.AddError(fmt.Errorf("unable to parse address '%s': %s", addr, err))
			continue
		}
		if u.Path == "" {
			u.Path = b.APIPrefix
		}

		wg.Add(1)
		go func(u *url.URL) {
			defer wg.Done()
			acc.AddError(b.gatherServer(u, acc))
		}(u)
	}

	wg.Wait()
	return nil
}

func (b *burrow) setDefaults() {
	if b.APIPrefix == "" {
		b.APIPrefix = defaultBurrowPrefix
	}
	if b.ConcurrentConnections < 1 {
		b.ConcurrentConnections = defaultConcurrentConnections
	}
	if time.Duration(b.ResponseTimeout) < time.Second {
		b.ResponseTimeout = config.Duration(defaultResponseTimeout)
	}
}

func (b *burrow) compileGlobs() error {
	var err error

	// compile glob patterns
	b.filterClusters, err = filter.NewIncludeExcludeFilter(b.ClustersInclude, b.ClustersExclude)
	if err != nil {
		return err
	}
	b.filterGroups, err = filter.NewIncludeExcludeFilter(b.GroupsInclude, b.GroupsExclude)
	if err != nil {
		return err
	}
	b.filterTopics, err = filter.NewIncludeExcludeFilter(b.TopicsInclude, b.TopicsExclude)
	if err != nil {
		return err
	}
	return nil
}

func (b *burrow) createClient() (*http.Client, error) {
	tlsCfg, err := b.ClientConfig.TLSConfig()
	if err != nil {
		return nil, err
	}

	timeout := time.Duration(b.ResponseTimeout)
	dialContext := net.Dialer{Timeout: timeout, DualStack: true}
	transport := http.Transport{
		DialContext:     dialContext.DialContext,
		TLSClientConfig: tlsCfg,
		// If b.ConcurrentConnections <= 1, then DefaultMaxIdleConnsPerHost is used (=2)
		MaxIdleConnsPerHost: b.ConcurrentConnections / 2,
		// If b.ConcurrentConnections == 0, then it is treated as "no limits"
		MaxConnsPerHost:       b.ConcurrentConnections,
		ResponseHeaderTimeout: timeout,
		IdleConnTimeout:       90 * time.Second,
	}

	client := &http.Client{
		Transport: &transport,
		Timeout:   timeout,
	}

	return client, nil
}

func (b *burrow) getResponse(u *url.URL) (*apiResponse, error) {
	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
	if err != nil {
		return nil, err
	}
	if b.Username != "" {
		req.SetBasicAuth(b.Username, b.Password)
{'fraction_non_alphanumeric': 0.12004483847956791, 'fraction_numerical': 0.006827677570569652, 'mean_word_length': 2.984571660576533, 'pattern_counts': {'":': 0, '<': 9, '<?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': '5193433', 'n_tokens_mistral': 3558, 'n_tokens_neox': 3263, 'n_words': 735}