Add ingest stage to provide synthetic workload for benchmarks #395

KalmanMeth · 2023-02-27T09:03:56Z

This PR introduces ingest_synthetic stage to provide a steady stream of flow logs for a configured number of simulated connections. This may be used to control the input flows to be able to perform benchmarking and memory usages of stages.
To configure the stage, add in the config file the following fields:

parameters:
- name: ingest_syn
  ingest:
    type: synthetic
    synthetic:
      connections: <nnn>
      batchMaxLen: <nnn>
      flowLogsPerMin: <nnn>

codecov · 2023-03-05T12:31:22Z

Codecov Report

Merging #395 (1db643a) into main (99a2598) will increase coverage by 0.29%.
The diff coverage is 88.75%.

@@            Coverage Diff             @@
##             main     #395      +/-   ##
==========================================
+ Coverage   64.09%   64.39%   +0.29%     
==========================================
  Files          92       94       +2     
  Lines        6501     6560      +59     
==========================================
+ Hits         4167     4224      +57     
- Misses       2094     2096       +2     
  Partials      240      240

Flag	Coverage Δ
unittests	`64.39% <88.75%> (+0.29%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
pkg/pipeline/pipeline_builder.go	`70.60% <0.00%> (-0.46%)`	⬇️
pkg/test/utils.go	`72.66% <ø> (-1.26%)`	⬇️
pkg/pipeline/utils/connections.go	`85.71% <85.71%> (ø)`
pkg/pipeline/ingest/ingest_synthetic.go	`93.75% <93.75%> (ø)`
pkg/config/config.go	`66.66% <100.00%> (+3.33%)`	⬆️

... and 1 file with indirect coverage changes

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

ronensc · 2023-03-27T09:51:56Z

pkg/pipeline/ingest/ingest_synthetic.go

+	metricsProcessed = operational.DefineMetric(
+		"ingest_synthetic_flows_processed",
+		"Number of metrics processed",
+		operational.TypeCounter,
+		"stage",
+	)


I think the variable name and the help string should be updated to reflect that it processes flow logs rather than metrics

ronensc · 2023-03-27T09:53:02Z

pkg/pipeline/ingest/ingest_synthetic.go

+// IngestSynthetic Ingest generates flow logs according to provided parameters
+func (ingestS *IngestSynthetic) Ingest(out chan<- config.GenericMap) {


Suggested change

// IngestSynthetic Ingest generates flow logs according to provided parameters

func (ingestS *IngestSynthetic) Ingest(out chan<- config.GenericMap) {

// Ingest generates flow logs according to provided parameters

func (ingestS *IngestSynthetic) Ingest(out chan<- config.GenericMap) {

ronensc · 2023-03-27T11:42:03Z

pkg/pipeline/ingest/ingest_synthetic.go

+	next := 0
+
+	// compute time interval between batches
+	ticker := time.NewTicker(time.Duration(int(time.Minute*time.Duration(ingestS.params.BatchMaxLen)) / ingestS.params.FlowLogsPerMin))


The computation of the interval between batches looks complex to me.
I don't think I understand it.
What does it mean to create a Duration from BatchMaxLen and multiply it by time.Minute?

time.Duration(ingestS.params.BatchMaxLen)

added a comment

ronensc · 2023-03-27T12:00:18Z

pkg/pipeline/ingest/ingest_synthetic_test.go

+
+	// Start collecting flows from the ingester and ensure we have the specified number of distinct connections
+	connectionMap := make(map[connection]int)
+	for i := 0; i < (3 * connections); i++ {


Is the hardcoded 3 the batchMaxLen? or just an arbitrary number to make sure we have multiple flow logs per connection?

It's just to have (many) more flow logs than connections, and to verify that we accumulate the proper number of connections with multiple flow logs per connection.

ronensc · 2023-03-28T06:58:51Z

pkg/pipeline/ingest/ingest_synthetic.go

+	jsonIngestSynthetic := api.IngestSynthetic{}
+	if params.Ingest != nil || params.Ingest.Synthetic != nil {
+		jsonIngestSynthetic = *params.Ingest.Synthetic
+	}
+	if jsonIngestSynthetic.Connections == 0 {
+		jsonIngestSynthetic.Connections = defaultConnections
+	}
+	if jsonIngestSynthetic.FlowLogsPerMin == 0 {
+		jsonIngestSynthetic.FlowLogsPerMin = defaultFlowLogsPerMin
+	}
+	if jsonIngestSynthetic.BatchMaxLen == 0 {
+		jsonIngestSynthetic.BatchMaxLen = defaultBatchLen
+	}


I wouldn't use "json" as part of the variable name of jsonIngestSynthetic because it's not limited to json only. Maybe "conf" could replace it?

ronensc

Thanks for addressing my feedback and adding the comments

ronensc · 2023-03-28T09:44:55Z

pkg/pipeline/ingest/ingest_synthetic.go

+type IngestSynthetic struct {
+	params           api.IngestSynthetic
+	exitChan         <-chan struct{}
+	metricsProcessed prometheus.Counter


Please rename metricsProcessed here as well

ronensc · 2023-03-28T09:51:02Z

pkg/pipeline/ingest/ingest_synthetic.go

+			for flowsLeft > 0 {
+				remainder := nLogs - next
+				if subBatchLen > remainder {
+					subBatchLen = remainder
+				}
+				log.Debugf("flowsLeft = %d, remainder = %d, subBatchLen = %d", flowsLeft, remainder, subBatchLen)
+				subBatch := flowLogs[next : next+subBatchLen]
+				ingestS.sendBatch(subBatch, out)
+				ingestS.metricsProcessed.Add(float64(subBatchLen))
+				flowsLeft -= subBatchLen
+				next += subBatchLen
+				if subBatchLen == remainder {
+					next = 0
+					subBatchLen = flowsLeft
+				}
+			}


I explored a different approach to achieve the goal of this loop:
https://github.com/ronensc/flowlogs-pipeline/blob/4266dc2e507e89c007a521d6eaeb7bc34de64abd/pkg/pipeline/ingest/ingest_synthetic.go#L73-L81

Please let me know if it is indeed equivalent and if you think it is clearer.
Since we loop over the sub-batch anyway in sendBatch() I incorporated it into the loop and removed sendBatch().

your suggestion is much simpler and cleaner. Done.

openshift-ci · 2023-03-28T11:49:56Z

[APPROVALNOTIFIER] This PR is APPROVED

Approval requirements bypassed by manually added approval.

This pull-request has been approved by:

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

KalmanMeth requested a review from ronensc March 1, 2023 11:09

KalmanMeth linked an issue Mar 6, 2023 that may be closed by this pull request

connection focused flowlogs simulator #372

Closed

KalmanMeth force-pushed the memory-measurement branch from 5b976fd to 9fb9930 Compare March 12, 2023 08:35

ronensc reviewed Mar 28, 2023

View reviewed changes

KalmanMeth added 7 commits March 28, 2023 12:00

move GenerateConnectionFlowEntries from test to utils

0ad41eb

added fields to GenerateConnectionEntries

0cd4f57

added infrastructure for ingest synthetic

a775571

added ingest synthetic

f16ee7f

added operational metric to count flow logs produced

a82b12d

added error message if metricsSettings missing

032255f

addressed reviewer comments

b00857b

KalmanMeth force-pushed the memory-measurement branch from 4d59cae to b00857b Compare March 28, 2023 09:01

ronensc reviewed Mar 28, 2023

View reviewed changes

simplified ingest loop

1db643a

ronensc approved these changes Mar 28, 2023

View reviewed changes

openshift-ci bot assigned ronensc Mar 28, 2023

openshift-ci bot added the lgtm label Mar 28, 2023

KalmanMeth added the approved label Mar 28, 2023

KalmanMeth merged commit a31a0c0 into netobserv:main Mar 28, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add ingest stage to provide synthetic workload for benchmarks #395

Add ingest stage to provide synthetic workload for benchmarks #395

KalmanMeth commented Feb 27, 2023 •

edited

Loading

codecov bot commented Mar 5, 2023 •

edited

Loading

ronensc Mar 27, 2023

ronensc Mar 27, 2023

KalmanMeth Mar 28, 2023

ronensc Mar 27, 2023

KalmanMeth Mar 28, 2023

ronensc Mar 27, 2023

KalmanMeth Mar 28, 2023

ronensc Mar 28, 2023

ronensc left a comment

ronensc Mar 28, 2023

ronensc Mar 28, 2023

KalmanMeth Mar 28, 2023

openshift-ci bot commented Mar 28, 2023

		// IngestSynthetic Ingest generates flow logs according to provided parameters
		func (ingestS *IngestSynthetic) Ingest(out chan<- config.GenericMap) {

Add ingest stage to provide synthetic workload for benchmarks #395

Add ingest stage to provide synthetic workload for benchmarks #395

Conversation

KalmanMeth commented Feb 27, 2023 • edited Loading

codecov bot commented Mar 5, 2023 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronensc left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

openshift-ci bot commented Mar 28, 2023

KalmanMeth commented Feb 27, 2023 •

edited

Loading

codecov bot commented Mar 5, 2023 •

edited

Loading