<commit_before>package broadcast

import (
	"context"
	"fmt"
	"net"
	"sort"
	"strings"

	"github.com/ansel1/merry"
	protov3 "github.com/go-graphite/protocol/carbonapi_v3_pb"
	"github.com/lomik/zapwriter"

	"github.com/go-graphite/carbonapi/limiter"
	"github.com/go-graphite/carbonapi/pathcache"
	utilctx "github.com/go-graphite/carbonapi/util/ctx"
	"github.com/go-graphite/carbonapi/zipper/helper"
	"github.com/go-graphite/carbonapi/zipper/types"

	"go.uber.org/zap"
)

type BroadcastGroup struct {
	limiter                   limiter.ServerLimiter
	groupName                 string
	timeout                   types.Timeouts
	backends                  []types.BackendServer
	servers                   []string
	maxMetricsPerRequest      int
	doMultipleRequestsIfSplit bool
	tldCacheDisabled          bool
	concurrencyLimit          int

	fetcher   types.Fetcher
	pathCache pathcache.PathCache
	logger    *zap.Logger
	dialer    *net.Dialer
}

type Option func(group *BroadcastGroup)

func WithLogger(logger *zap.Logger) Option {
	return func(bg *BroadcastGroup) {
		bg.logger = logger
	}
}

func WithGroupName(name string) Option {
	return func(bg *BroadcastGroup) {
		bg.groupName = name
	}
}

func WithSplitMultipleRequests(multiRequests bool) Option {
	return func(bg *BroadcastGroup) {
		bg.doMultipleRequestsIfSplit = multiRequests
		if multiRequests {
			bg.fetcher = bg.doMultiFetch
		} else {
			bg.fetcher = bg.doSingleFetch
		}
	}
}

func WithBackends(backends []types.BackendServer) Option {
	return func(bg *BroadcastGroup) {
		serverNames := make([]string, 0, len(backends))
		for _, b := range backends {
			serverNames = append(serverNames, b.Name())
		}
		bg.backends = backends
		bg.servers = serverNames
	}
}

func WithPathCache(expireDelaySec int32) Option {
	return func(bg *BroadcastGroup) {
		bg.pathCache = pathcache.NewPathCache(expireDelaySec)
	}
}

func WithLimiter(concurrencyLimit int) Option {
	return func(bg *BroadcastGroup) {
		bg.concurrencyLimit = concurrencyLimit
	}
}

func WithMaxMetricsPerRequest(maxMetricsPerRequest int) Option {
	return func(bg *BroadcastGroup) {
		bg.maxMetricsPerRequest = maxMetricsPerRequest
	}
}

func WithTLDCache(enableTLDCache bool) Option {
	return func(bg *BroadcastGroup) {
		bg.tldCacheDisabled = !enableTLDCache
	}
}

func WithTimeouts(timeouts types.Timeouts) Option {
	return func(bg *BroadcastGroup) {
		bg.timeout = timeouts
	}
}

func WithDialer(dialer *net.Dialer) Option {
	return func(bg *BroadcastGroup) {
		bg.dialer = dialer
	}
}

func New(opts ...Option) (*BroadcastGroup, merry.Error) {
	bg := &BroadcastGroup{
		limiter: limiter.NoopLimiter{},
	}

	for _, opt := range opts {
		opt(bg)
	}

	if bg.logger == nil {
		logger := zapwriter.Logger("init")
		logger.Fatal("failed to initialize backend")
	}

	bg.logger = bg.logger.With(zap.String("type", "broadcastGroup"), zap.String("groupName", bg.groupName))

	if len(bg.backends) == 0 {
		return nil, types.ErrNoServersSpecified
	}

	if bg.concurrencyLimit != 0 {
		bg.limiter = limiter.NewServerLimiter(bg.servers, bg.concurrencyLimit)
	}

	return bg, nil
}

func (bg *BroadcastGroup) Children() []types.BackendServer {
	return bg.backends
}

func (bg *BroadcastGroup) SetDoMultipleRequestIfSplit(v bool) {
	bg.doMultipleRequestsIfSplit = v
	if v {
		bg.fetcher = bg.doMultiFetch
	} else {
		bg.fetcher = bg.doSingleFetch
	}
}

func NewBroadcastGroup(logger *zap.Logger, groupName string, doMultipleRequestsIfSplit bool, servers []types.BackendServer, expireDelaySec int32, concurrencyLimit, maxBatchSize int, timeouts types.Timeouts, tldCacheDisabled bool) (*BroadcastGroup, merry.Error) {
	return New(
		WithLogger(logger),
		WithGroupName(groupName),
		WithSplitMultipleRequests(doMultipleRequestsIfSplit),
		WithBackends(servers),
		WithPathCache(expireDelaySec),
		WithLimiter(concurrencyLimit),
		WithMaxMetricsPerRequest(maxBatchSize),
		WithTimeouts(timeouts),
		WithTLDCache(!tldCacheDisabled),
	)
}

func (bg BroadcastGroup) Name() string {
	return bg.groupName
}

func (bg BroadcastGroup) Backends() []string {
	return bg.servers
}

func (bg *BroadcastGroup) filterServersByTLD(requests []string, backends []types.BackendServer) []types.BackendServer {
	// do not check TLDs if internal routing cache is disabled
	if bg.tldCacheDisabled {
		return backends
	}

	tldBackends := make(map[types.BackendServer]bool)
	for _, request := range requests {
		// TODO(Civil): Tags: improve logic
		if strings.HasPrefix(request, "seriesByTag") {
			return backends
		}
		idx := strings.Index(request, ".")
		if idx > 0 {
			request = request[:idx]
		}
		if cachedBackends, ok := bg.pathCache.Get(request); ok && len(backends) > 0 {
			for _, cachedBackend := range cachedBackends {
				tldBackends[cachedBackend] = true
			}
		}
	}

	var filteredBackends []types.BackendServer
	for _, k := range backends {
		if tldBackends[k] {
			filteredBackends = append(filteredBackends, k)
		}
	}

	if len(filteredBackends) == 0 {
		return backends
	}

	return filteredBackends
}

func (bg BroadcastGroup) MaxMetricsPerRequest() int {
	return bg.maxMetricsPerRequest
}

func (bg *BroadcastGroup) doMultiFetch(ctx context.Context, logger *zap.Logger, backend types.BackendServer, reqs interface{}, resCh chan types.ServerFetcherResponse) {
	logger = logger.With(zap.Bool("multi_fetch", true))
	request, ok := reqs.(*protov3.MultiFetchRequest)
	if !ok {
		logger.Fatal("unhandled error in doMultiFetch",
			zap.Stack("stack"),
			zap.String("got_type", fmt.Sprintf("%T", reqs)),
			zap.String("expected_type", fmt.Sprintf("%T", request)),
		)
	}

	requests, err := bg.splitRequest(ctx, request, backend)
	if len(requests) == 0 && err != nil {
		response := types.NewServerFetchResponse()
		response.Server = backend.Name()
		response.AddError(err)
		resCh <- response
		return
	}

	for _, req := range requests {
		go func(req *protov3.MultiFetchRequest) {
			logger = logger.With(zap.String("backend_name", backend.Name()))
			logger.Debug("waiting for slot",
				zap.Int("max_connections", bg.limiter.Capacity()),
			)

			response := types.NewServerFetchResponse()
			response.Server = backend.Name()

			if err := bg.limiter.Enter(ctx, backend.Name()); err != nil {
				logger.Debug("timeout waiting for a slot")
				resCh <- response.NonFatalError(merry.Prepend(err, "timeout waiting for slot"))
				return
			}

			logger.Debug("got slot")
			defer bg.limiter.Leave(ctx, backend.Name())

			// uuid := util.GetUUID(ctx)
			var err merry.Error
			logger.Debug("sending request")
			response.Response, response.Stats, err = backend.Fetch(ctx, req)
			response.AddError(err)
			logger.Debug("got response",
				zap.Int("metrics_in_response", len(response.Response.Metrics)),
				zap.Int("errors_count", len(response.Err)),
				zap.Int64("timeouts_count", response.Stats.Timeouts),
				zap.Int64("render_requests_count", response.Stats.RenderRequests),
				zap.Int64("render_errors_count", response.Stats.RenderErrors),
				zap.Int64("render_timeouts_count", response.Stats.RenderTimeouts),
				zap.Int64("zipper_requests_count", response.Stats.ZipperRequests),
				zap.Int64("total_metric_count", response.Stats.TotalMetricsCount),
				zap.Int("servers_count", len(response.Stats.Servers)),
				zap.Int("failed_servers_count", len(response.Stats.FailedServers)),
			)

			resCh <- response
		}(req)
	}

}

func (bg *BroadcastGroup) doSingleFetch(ctx context.Context, logger *zap.Logger, backend types.BackendServer, reqs interface{}, resCh chan types.ServerFetcherResponse) {
	logger = logger.With(zap.Bool("multi_fetch", false))
	request, ok := reqs.(*protov3.MultiFetchRequest)
	if !ok {
		logger.Fatal("unhandled error in doSingleFetch",
			zap.Stack("stack"),
			zap.String("got_type", fmt.Sprintf("%T", reqs)),
			zap.String("expected_type", fmt.Sprintf("%T", request)),
		)
	}

	//TODO(Civil): migrate limiter to merry
	requests, splitErr := bg.splitRequest(ctx, request, backend)
	if len(requests) == 0 && splitErr != nil {
		response := types.NewServerFetchResponse()
		response.Server = backend.Name()
		response.AddError(splitErr)
		resCh <- response
		return
	}

	logger = logger.With(zap.String("backend_name", backend.Name()))
	logger.Debug("waiting for slot",
		zap.Int("max_connections", bg.limiter.Capacity()),
	)

	response := types.NewServerFetchResponse()
	response.Server = backend.Name()

	if err := bg.limiter.Enter(ctx, backend.Name()); err != nil {
		logger.Debug("timeout waiting for a slot")
		resCh <- response.NonFatalError(merry.Prepend(err, "timeout waiting for slot"))
		return
	}

	logger.Debug("got slot")
	defer bg.limiter.Leave(ctx, backend.Name())

	// uuid := util.GetUUID(ctx)
	var err merry.Error
	for _, req := range requests {
		logger.Debug("sending request")
		r := types.NewServerFetchResponse()
		r.Response, r.Stats, err = backend.Fetch(ctx, req)
		r.AddError(err)
		logger.Debug("got response",
			zap.Int("metrics_in_response", len(r.Response.Metrics)),
			zap.Int("errors_count", len(r.Err)),
			zap.Int64("timeouts_count", r.Stats.Timeouts),
			zap.Int64("render_requests_count", r.Stats.RenderRequests),
			zap.Int64("render_errors_count", r.Stats.RenderErrors),
			zap.Int64("render_timeouts_count", r.Stats.RenderTimeouts),
			zap.Int64("zipper_requests_count", r.Stats.ZipperRequests),
			zap.Int64("total_metric_count", r.Stats.TotalMetricsCount),
			zap.Int("servers_count", len(r.Stats.Servers)),
			zap.Int("failed_servers_count", len(r.Stats.FailedServers)),
		)
		_ = response.Merge(r)
	}
	logger.Debug("got response (after merge)",
		zap.Int("metrics_in_response", len(response.Response.Metrics)),
		zap.Int("errors_count", len(response.Err)),
		zap.Int64("timeouts_count", response.Stats.Timeouts),
		zap.Int64("render_requests_count", response.Stats.RenderRequests),
		zap.Int64("render_errors_count", response.Stats.RenderErrors),
		zap.Int64("render_timeouts_count", response.Stats.RenderTimeouts),
		zap.Int64("zipper_requests_count", response.Stats.ZipperRequests),
		zap.Int64("total_metric_count", response.Stats.TotalMetricsCount),
		zap.Int("servers_count", len(response.Stats.Servers)),
		zap.Int("failed_servers_count", len(response.Stats.FailedServers)),
	)

	resCh <- response
}

func (bg *BroadcastGroup) splitRequest(ctx context.Context, request *protov3.MultiFetchRequest, backend types.BackendServer) ([]*protov3.MultiFetchRequest, merry.Error) {
	if backend.MaxMetricsPerRequest() == 0 {
		return []*protov3.MultiFetchRequest{request}, nil
	}

	var requests []*protov3.MultiFetchRequest<commit_msg>Fix panic in new debug logs when some responses have failed<commit_after>package broadcast

import (
	"context"
	"fmt"
	"net"
	"sort"
	"strings"

	"github.com/ansel1/merry"
	protov3 "github.com/go-graphite/protocol/carbonapi_v3_pb"
	"github.com/lomik/zapwriter"

	"github.com/go-graphite/carbonapi/limiter"
	"github.com/go-graphite/carbonapi/pathcache"
	utilctx "github.com/go-graphite/carbonapi/util/ctx"
	"github.com/go-graphite/carbonapi/zipper/helper"
	"github.com/go-graphite/carbonapi/zipper/types"

	"go.uber.org/zap"
)

type BroadcastGroup struct {
	limiter                   limiter.ServerLimiter
	groupName                 string
	timeout                   types.Timeouts
	backends                  []types.BackendServer
	servers                   []string
	maxMetricsPerRequest      int
	doMultipleRequestsIfSplit bool
	tldCacheDisabled          bool
	concurrencyLimit          int

	fetcher   types.Fetcher
	pathCache pathcache.PathCache
	logger    *zap.Logger
	dialer    *net.Dialer
}

type Option func(group *BroadcastGroup)

func WithLogger(logger *zap.Logger) Option {
	return func(bg *BroadcastGroup) {
		bg.logger = logger
	}
}

func WithGroupName(name string) Option {
	return func(bg *BroadcastGroup) {
		bg.groupName = name
	}
}

func WithSplitMultipleRequests(multiRequests bool) Option {
	return func(bg *BroadcastGroup) {
		bg.doMultipleRequestsIfSplit = multiRequests
		if multiRequests {
			bg.fetcher = bg.doMultiFetch
		} else {
			bg.fetcher = bg.doSingleFetch
		}
	}
}

func WithBackends(backends []types.BackendServer) Option {
	return func(bg *BroadcastGroup) {
		serverNames := make([]string, 0, len(backends))
		for _, b := range backends {
			serverNames = append(serverNames, b.Name())
		}
		bg.backends = backends
		bg.servers = serverNames
	}
}

func WithPathCache(expireDelaySec int32) Option {
	return func(bg *BroadcastGroup) {
		bg.pathCache = pathcache.NewPathCache(expireDelaySec)
	}
}

func WithLimiter(concurrencyLimit int) Option {
	return func(bg *BroadcastGroup) {
		bg.concurrencyLimit = concurrencyLimit
	}
}

func WithMaxMetricsPerRequest(maxMetricsPerRequest int) Option {
	return func(bg *BroadcastGroup) {
		bg.maxMetricsPerRequest = maxMetricsPerRequest
	}
}

func WithTLDCache(enableTLDCache bool) Option {
	return func(bg *BroadcastGroup) {
		bg.tldCacheDisabled = !enableTLDCache
	}
}

func WithTimeouts(timeouts types.Timeouts) Option {
	return func(bg *BroadcastGroup) {
		bg.timeout = timeouts
	}
}

func WithDialer(dialer *net.Dialer) Option {
	return func(bg *BroadcastGroup) {
		bg.dialer = dialer
	}
}

func New(opts ...Option) (*BroadcastGroup, merry.Error) {
	bg := &BroadcastGroup{
		limiter: limiter.NoopLimiter{},
	}

	for _, opt := range opts {
		opt(bg)
	}

	if bg.logger == nil {
		logger := zapwriter.Logger("init")
		logger.Fatal("failed to initialize backend")
	}

	bg.logger = bg.logger.With(zap.String("type", "broadcastGroup"), zap.String("groupName", bg.groupName))

	if len(bg.backends) == 0 {
		return nil, types.ErrNoServersSpecified
	}

	if bg.concurrencyLimit != 0 {
		bg.limiter = limiter.NewServerLimiter(bg.servers, bg.concurrencyLimit)
	}

	return bg, nil
}

func (bg *BroadcastGroup) Children() []types.BackendServer {
	return bg.backends
}

func (bg *BroadcastGroup) SetDoMultipleRequestIfSplit(v bool) {
	bg.doMultipleRequestsIfSplit = v
	if v {
		bg.fetcher = bg.doMultiFetch
	} else {
		bg.fetcher = bg.doSingleFetch
	}
}

func NewBroadcastGroup(logger *zap.Logger, groupName string, doMultipleRequestsIfSplit bool, servers []types.BackendServer, expireDelaySec int32, concurrencyLimit, maxBatchSize int, timeouts types.Timeouts, tldCacheDisabled bool) (*BroadcastGroup, merry.Error) {
	return New(
		WithLogger(logger),
		WithGroupName(groupName),
		WithSplitMultipleRequests(doMultipleRequestsIfSplit),
		WithBackends(servers),
		WithPathCache(expireDelaySec),
		WithLimiter(concurrencyLimit),
		WithMaxMetricsPerRequest(maxBatchSize),
		WithTimeouts(timeouts),
		WithTLDCache(!tldCacheDisabled),
	)
}

func (bg BroadcastGroup) Name() string {
	return bg.groupName
}

func (bg BroadcastGroup) Backends() []string {
	return bg.servers
}

func (bg *BroadcastGroup) filterServersByTLD(requests []string, backends []types.BackendServer) []types.BackendServer {
	// do not check TLDs if internal routing cache is disabled
	if bg.tldCacheDisabled {
		return backends
	}

	tldBackends := make(map[types.BackendServer]bool)
	for _, request := range requests {
		// TODO(Civil): Tags: improve logic
		if strings.HasPrefix(request, "seriesByTag") {
			return backends
		}
		idx := strings.Index(request, ".")
		if idx > 0 {
			request = request[:idx]
		}
		if cachedBackends, ok := bg.pathCache.Get(request); ok && len(backends) > 0 {
			for _, cachedBackend := range cachedBackends {
				tldBackends[cachedBackend] = true
			}
		}
	}

	var filteredBackends []types.BackendServer
	for _, k := range backends {
		if tldBackends[k] {
			filteredBackends = append(filteredBackends, k)
		}
	}

	if len(filteredBackends) == 0 {
		return backends
	}

	return filteredBackends
}

func (bg BroadcastGroup) MaxMetricsPerRequest() int {
	return bg.maxMetricsPerRequest
}

func (bg *BroadcastGroup) doMultiFetch(ctx context.Context, logger *zap.Logger, backend types.BackendServer, reqs interface{}, resCh chan types.ServerFetcherResponse) {
	logger = logger.With(zap.Bool("multi_fetch", true))
	request, ok := reqs.(*protov3.MultiFetchRequest)
	if !ok {
		logger.Fatal("unhandled error in doMultiFetch",
			zap.Stack("stack"),
			zap.String("got_type", fmt.Sprintf("%T", reqs)),
			zap.String("expected_type", fmt.Sprintf("%T", request)),
		)
	}

	requests, err := bg.splitRequest(ctx, request, backend)
	if len(requests) == 0 && err != nil {
		response := types.NewServerFetchResponse()
		response.Server = backend.Name()
		response.AddError(err)
		resCh <- response
		return
	}

	for _, req := range requests {
		go func(req *protov3.MultiFetchRequest) {
			logger = logger.With(zap.String("backend_name", backend.Name()))
			logger.Debug("waiting for slot",
				zap.Int("max_connections", bg.limiter.Capacity()),
			)

			response := types.NewServerFetchResponse()
			response.Server = backend.Name()

			if err := bg.limiter.Enter(ctx, backend.Name()); err != nil {
				logger.Debug("timeout waiting for a slot")
				resCh <- response.NonFatalError(merry.Prepend(err, "timeout waiting for slot"))
				return
			}

			logger.Debug("got slot")
			defer bg.limiter.Leave(ctx, backend.Name())

			// uuid := util.GetUUID(ctx)
			var err merry.Error
			logger.Debug("sending request")
			response.Response, response.Stats, err = backend.Fetch(ctx, req)
			response.AddError(err)
			if response.Response != nil && response.Stats != nil {
				logger.Debug("got response",
					zap.Int("metrics_in_response", len(response.Response.Metrics)),
					zap.Int("errors_count", len(response.Err)),
					zap.Int64("timeouts_count", response.Stats.Timeouts),
					zap.Int64("render_requests_count", response.Stats.RenderRequests),
					zap.Int64("render_errors_count", response.Stats.RenderErrors),
					zap.Int64("render_timeouts_count", response.Stats.RenderTimeouts),
					zap.Int64("zipper_requests_count", response.Stats.ZipperRequests),
					zap.Int64("total_metric_count", response.Stats.TotalMetricsCount),
					zap.Int("servers_count", len(response.Stats.Servers)),
					zap.Int("failed_servers_count", len(response.Stats.FailedServers)),
				)
			} else {
				logger.Debug("got response",
					zap.Bool("response_is_nil", response.Response == nil),
					zap.Bool("stats_is_nil", response.Stats == nil),
					zap.Any("err", err),
				)
			}

			resCh <- response
		}(req)
	}

}

func (bg *BroadcastGroup) doSingleFetch(ctx context.Context, logger *zap.Logger, backend types.BackendServer, reqs interface{}, resCh chan types.ServerFetcherResponse) {
	logger = logger.With(zap.Bool("multi_fetch", false))
	request, ok := reqs.(*protov3.MultiFetchRequest)
	if !ok {
		logger.Fatal("unhandled error in doSingleFetch",
			zap.Stack("stack"),
			zap.String("got_type", fmt.Sprintf("%T", reqs)),
			zap.String("expected_type", fmt.Sprintf("%T", request)),
		)
	}

	//TODO(Civil): migrate limiter to merry
	requests, splitErr := bg.splitRequest(ctx, request, backend)
	if len(requests) == 0 && splitErr != nil {
		response := types.NewServerFetchResponse()
		response.Server = backend.Name()
		response.AddError(splitErr)
		resCh <- response
		return
	}

	logger = logger.With(zap.String("backend_name", backend.Name()))
	logger.Debug("waiting for slot",
		zap.Int("max_connections", bg.limiter.Capacity()),
	)

	response := types.NewServerFetchResponse()
	response.Server = backend.Name()

	if err := bg.limiter.Enter(ctx, backend.Name()); err != nil {
		logger.Debug("timeout waiting for a slot")
		resCh <- response.NonFatalError(merry.Prepend(err, "timeout waiting for slot"))
		return
	}

	logger.Debug("got slot")
	defer bg.limiter.Leave(ctx, backend.Name())

	// uuid := util.GetUUID(ctx)
	var err merry.Error
	for _, req := range requests {
		logger.Debug("sending request")
		r := types.NewServerFetchResponse()
		r.Response, r.Stats, err = backend.Fetch(ctx, req)
		r.AddError(err)
		if r.Stats != nil && r.Response != nil {
			logger.Debug("got response",
				zap.Int("metrics_in_response", len(r.Response.Metrics)),
				zap.Int("errors_count", len(r.Err)),
				zap.Int64("timeouts_count", r.Stats.Timeouts),
				zap.Int64("render_requests_count", r.Stats.RenderRequests),
				zap.Int64("render_errors_count", r.Stats.RenderErrors),
				zap.Int64("render_timeouts_count", r.Stats.RenderTimeouts),
				zap.Int64("zipper_requests_count", r.Stats.ZipperRequests),
				zap.Int64("total_metric_count", r.Stats.TotalMetricsCount),
				zap.Int("servers_count", len(r.Stats.Servers)),
				zap.Int("failed_servers_count", len(r.Stats.FailedServers)),
			)
		} else {
			logger.Debug("got response",
				zap.Bool("response_is_nil", r.Response == nil),
				zap.Bool("stats_is_nil", r.Stats == nil),
				zap.Any("err", err),
			)
		}
		_ = response.Merge(r)
	}
	logger.Debug("got response (after merge)",
		zap.Int("metrics_in_response", len(response.Response.Metrics)),
		zap.Int("errors_count", len(response.Err)),
		zap.Int64("timeouts_count", response.Stats.Timeouts),
		zap.Int64("render_requests_count", response.Stats.RenderRequests),
		zap.Int64("render_errors_count", response.Stats.RenderErrors),
		zap.Int64("render_timeouts_count", response.Stats.RenderTimeouts),
		zap.Int64("zipper_requests_count", response.Stats.ZipperRequests),
		zap.Int64("total_metric_count", response.Stats.TotalMetricsCount),
		zap.Int("servers_count", len(response.Stats.Servers)),
		zap.Int("failed_servers_count", len(response.Stats.FailedServers)),
	)

	resCh <- response
}

func (bg *BroadcastGroup) splitRequest(ctx context.Context, request *protov3.MultiFetchRequest, backend types.BackendServer) ([]*protov3.MultiFetchRequest, merry.Error) {
	if backend.MaxMetricsPerRequest() == 0 {
		return []*protov3.MultiFetchRequest{request}, nil
	}

	var requests []*protov3.MultiFetchRequest
{'fraction_non_alphanumeric': 0.132392535512023, 'fraction_numerical': 0.005477671525392257, 'mean_word_length': 4.582534335320031, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 7, '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': '3206002', 'n_tokens_mistral': 7835, 'n_tokens_neox': 7244, 'n_words': 1369}