16114132341

Committed 07 Jul 2025 10:09AM UTC coverage: 76.124% (-0.07%) from 76.193%

Build # 16114132341

Build Type

Pull #3430

github

Committed by

orestisfl

Commit Message

Add mocks

Pull Request Pull Request #3430: Add OpenTelemetry

Run Details

147 of 205 new or added lines in 7 files covered. (71.71%)

2 existing lines in 1 file now uncovered.

9463 of 12431 relevant lines covered (76.12%)

16.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/internal/flavors/posture.go

// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package flavors

import (
        "context"
        "fmt"

        "github.com/elastic/beats/v7/libbeat/beat"
        agentconfig "github.com/elastic/elastic-agent-libs/config"
        "github.com/elastic/elastic-agent-libs/logp"

        "github.com/elastic/cloudbeat/internal/config"
        "github.com/elastic/cloudbeat/internal/flavors/benchmark"
        "github.com/elastic/cloudbeat/internal/flavors/benchmark/builder"
        "github.com/elastic/cloudbeat/internal/infra/clog"
        "github.com/elastic/cloudbeat/internal/infra/observability"
        _ "github.com/elastic/cloudbeat/internal/processor" // Add cloudbeat default processors.
)

type posture struct {
        flavorBase
        benchmark builder.Benchmark
}

// NewPosture creates an instance of posture.
func NewPosture(b *beat.Beat, agentConfig *agentconfig.C) (beat.Beater, error) {
        cfg, err := config.New(agentConfig)
        if err != nil {
                return nil, fmt.Errorf("error reading config file: %w", err)
        }
        return newPostureFromCfg(b, cfg)
}

// NewPosture creates an instance of posture.
func newPostureFromCfg(b *beat.Beat, cfg *config.Config) (*posture, error) {
        log := clog.NewLogger("posture")
        log.Info("Config initiated with cycle period of ", cfg.Period)
        ctx, cancel := context.WithCancel(context.Background())

        ctx, err := observability.SetUpOtel(ctx, log.Logger)
        if err != nil {
                log.Errorw("failed to set up otel", logp.Error(err))
        }

        strategy, err := benchmark.GetStrategy(cfg, log)
        if err != nil {
                cancel()
                return nil, err
        }

        log.Infof("Creating benchmark %T", strategy)
        bench, err := strategy.NewBenchmark(ctx, log, cfg)
        if err != nil {
                cancel()
                return nil, err
        }

        err = ensureHostProcessor(log, cfg)
        if err != nil {
                cancel()
                return nil, err
        }

        client, err := NewClient(b.Publisher, cfg.Processors)
        if err != nil {
                cancel()
                return nil, fmt.Errorf("failed to init client: %w", err)
        }
        log.Infof("posture configured %d processors", len(cfg.Processors))

        publisher := NewPublisher(log, flushInterval, eventsThreshold, client)

        return &posture{
                flavorBase: flavorBase{
                        ctx:       ctx,
                        cancel:    cancel,
                        publisher: publisher,
                        config:    cfg,
                        log:       log,
                        client:    client,
                },
                benchmark: bench,
        }, nil
}

// Run starts posture.
func (bt *posture) Run(*beat.Beat) error {
        bt.log.Info("posture is running! Hit CTRL-C to stop it")
        eventsCh, err := bt.benchmark.Run(bt.ctx)
        if err != nil {
                return err
        }

        bt.publisher.HandleEvents(bt.ctx, eventsCh)
        bt.log.Warn("Posture has finished running")
        return nil
}

// Stop stops posture.
func (bt *posture) Stop() {
        defer bt.cancel() // context cancellation should be the last action
        bt.benchmark.Stop()

        if err := bt.client.Close(); err != nil {
                bt.log.Fatal("Cannot close client", err)
        }

        if err := observability.ShutdownOtel(bt.ctx); err != nil {
                bt.log.Warnw("Failed to shutdown otel", logp.Error(err))
        }
}

// ensureAdditionalProcessors modifies cfg.Processors list to ensure 'host'
// processor is present for K8s and EKS benchmarks.
func ensureHostProcessor(log *clog.Logger, cfg *config.Config) error {
        if cfg.Benchmark != config.CIS_EKS && cfg.Benchmark != config.CIS_K8S {
                return nil
        }
        log.Info("Adding host processor config")
        hostProcessor, err := agentconfig.NewConfigFrom("add_host_metadata: ~")
        if err != nil {
                return err
        }
        cfg.Processors = append(cfg.Processors, hostProcessor)
        return nil
}

1	// Licensed to Elasticsearch B.V. under one or more contributor
2	// license agreements. See the NOTICE file distributed with
3	// this work for additional information regarding copyright
4	// ownership. Elasticsearch B.V. licenses this file to you under
5	// the Apache License, Version 2.0 (the "License"); you may
6	// not use this file except in compliance with the License.
7	// You may obtain a copy of the License at
8	//
9	// http://www.apache.org/licenses/LICENSE-2.0
10	//
11	// Unless required by applicable law or agreed to in writing,
12	// software distributed under the License is distributed on an
13	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14	// KIND, either express or implied. See the License for the
15	// specific language governing permissions and limitations
16	// under the License.
17
18	package flavors
19
20	import (
21	"context"
22	"fmt"
23
24	"github.com/elastic/beats/v7/libbeat/beat"
25	agentconfig "github.com/elastic/elastic-agent-libs/config"
26	"github.com/elastic/elastic-agent-libs/logp"
27
28	"github.com/elastic/cloudbeat/internal/config"
29	"github.com/elastic/cloudbeat/internal/flavors/benchmark"
30	"github.com/elastic/cloudbeat/internal/flavors/benchmark/builder"
31	"github.com/elastic/cloudbeat/internal/infra/clog"
32	"github.com/elastic/cloudbeat/internal/infra/observability"
33	_ "github.com/elastic/cloudbeat/internal/processor" // Add cloudbeat default processors.
34	)
35
36	type posture struct {
37	flavorBase
38	benchmark builder.Benchmark
39	}
40
41	// NewPosture creates an instance of posture.
42	func NewPosture(b beat.Beat, agentConfig agentconfig.C) (beat.Beater, error) {	×
43	cfg, err := config.New(agentConfig)	×
44	if err != nil {	×
45	return nil, fmt.Errorf("error reading config file: %w", err)	×
46	}	×
47	return newPostureFromCfg(b, cfg)	×
48	}
49
50	// NewPosture creates an instance of posture.
51	func newPostureFromCfg(b beat.Beat, cfg config.Config) (*posture, error) {	×
52	log := clog.NewLogger("posture")	×
53	log.Info("Config initiated with cycle period of ", cfg.Period)	×
54	ctx, cancel := context.WithCancel(context.Background())	×
55		×
NEW 56	ctx, err := observability.SetUpOtel(ctx, log.Logger)	×
NEW 57	if err != nil {	×
NEW 58	log.Errorw("failed to set up otel", logp.Error(err))	×
NEW 59	}	×
60
61	strategy, err := benchmark.GetStrategy(cfg, log)	×
62	if err != nil {	×
63	cancel()	×
64	return nil, err	×
65	}	×
66
67	log.Infof("Creating benchmark %T", strategy)	×
68	bench, err := strategy.NewBenchmark(ctx, log, cfg)	×
69	if err != nil {	×
70	cancel()	×
71	return nil, err	×
72	}	×
73
74	err = ensureHostProcessor(log, cfg)	×
75	if err != nil {	×
76	cancel()	×
77	return nil, err	×
78	}	×
79
80	client, err := NewClient(b.Publisher, cfg.Processors)	×
81	if err != nil {	×
82	cancel()	×
83	return nil, fmt.Errorf("failed to init client: %w", err)	×
84	}	×
85	log.Infof("posture configured %d processors", len(cfg.Processors))	×
86		×
87	publisher := NewPublisher(log, flushInterval, eventsThreshold, client)	×
88		×
89	return &posture{	×
90	flavorBase: flavorBase{	×
91	ctx: ctx,	×
92	cancel: cancel,	×
93	publisher: publisher,	×
94	config: cfg,	×
95	log: log,	×
96	client: client,	×
97	},	×
98	benchmark: bench,	×
99	}, nil	×
100	}
101
102	// Run starts posture.
103	func (bt posture) Run(beat.Beat) error {	×
104	bt.log.Info("posture is running! Hit CTRL-C to stop it")	×
105	eventsCh, err := bt.benchmark.Run(bt.ctx)	×
106	if err != nil {	×
107	return err	×
108	}	×
109
110	bt.publisher.HandleEvents(bt.ctx, eventsCh)	×
111	bt.log.Warn("Posture has finished running")	×
112	return nil	×
113	}
114
115	// Stop stops posture.
116	func (bt *posture) Stop() {	×
NEW 117	defer bt.cancel() // context cancellation should be the last action	×
118	bt.benchmark.Stop()	×
119		×
120	if err := bt.client.Close(); err != nil {	×
121	bt.log.Fatal("Cannot close client", err)	×
122	}	×
123
NEW 124	if err := observability.ShutdownOtel(bt.ctx); err != nil {	×
NEW 125	bt.log.Warnw("Failed to shutdown otel", logp.Error(err))	×
NEW 126	}	×
127	}
128
129	// ensureAdditionalProcessors modifies cfg.Processors list to ensure 'host'
130	// processor is present for K8s and EKS benchmarks.
131	func ensureHostProcessor(log clog.Logger, cfg config.Config) error {	×
132	if cfg.Benchmark != config.CIS_EKS && cfg.Benchmark != config.CIS_K8S {	×
133	return nil	×
134	}	×
135	log.Info("Adding host processor config")	×
136	hostProcessor, err := agentconfig.NewConfigFrom("add_host_metadata: ~")	×
137	if err != nil {	×
138	return err	×
139	}	×
140	cfg.Processors = append(cfg.Processors, hostProcessor)	×
141	return nil	×
142	}

elastic / cloudbeat / 16114132341

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous