@@ -2,21 +2,15 @@ package reports
22
33import (
44 "bufio"
5- "bytes"
6- "encoding/binary"
75 "encoding/json"
8- "errors"
96 "fmt"
107 "io"
118 "os"
129 "path/filepath"
13- "strings"
14- "sync"
15- "time"
1610
17- "github.com/go-resty/resty/v2"
1811 "github.com/google/uuid"
1912 "github.com/rs/zerolog/log"
13+ "golang.org/x/sync/errgroup"
2014)
2115
2216// FileSystem interface and implementations
@@ -47,6 +41,8 @@ type aggregateOptions struct {
4741 splunkURL string
4842 splunkToken string
4943 splunkEvent string
44+ dxURL string
45+ dxToken string
5046 branchName string
5147 baseSha string
5248 headSha string
@@ -66,6 +62,7 @@ func WithReportID(reportID string) AggregateOption {
6662}
6763
6864// WithSplunk also sends the aggregation to a Splunk instance as events.
65+ // https://www.splunk.com/
6966func WithSplunk (url , token string , event string ) AggregateOption {
7067 return func (opts * aggregateOptions ) {
7168 opts .splunkURL = url
@@ -74,6 +71,15 @@ func WithSplunk(url, token string, event string) AggregateOption {
7471 }
7572}
7673
74+ // WithDX also sends the aggregation to DX.
75+ // https://getdx.com/
76+ func WithDX (url , token string ) AggregateOption {
77+ return func (opts * aggregateOptions ) {
78+ opts .dxURL = url
79+ opts .dxToken = token
80+ }
81+ }
82+
7783func WithBranchName (branchName string ) AggregateOption {
7884 return func (opts * aggregateOptions ) {
7985 opts .branchName = branchName
@@ -444,7 +450,6 @@ func aggregate(reportChan <-chan *TestReport, errChan <-chan error, opts *aggreg
444450 testMap = make (map [string ]TestResult )
445451 excludedTests = map [string ]struct {}{}
446452 selectedTests = map [string ]struct {}{}
447- sendToSplunk = opts .splunkURL != ""
448453 )
449454
450455 for report := range reportChan {
@@ -497,222 +502,18 @@ func aggregate(reportChan <-chan *TestReport, errChan <-chan error, opts *aggreg
497502 fullReport .Results = aggregatedResults
498503 fullReport .GenerateSummaryData ()
499504
500- if sendToSplunk {
501- err = sendDataToSplunk (opts , * fullReport )
502- if err != nil {
503- return fullReport , fmt .Errorf ("error sending data to Splunk: %w" , err )
504- }
505- }
506- return fullReport , err
507- }
508-
509- // sendDataToSplunk sends a truncated TestReport and each individual TestResults to Splunk as events
510- func sendDataToSplunk (opts * aggregateOptions , report TestReport ) error {
511- start := time .Now ()
512- results := report .Results
513- report .Results = nil // Don't send results to Splunk, doing that individually
514- // Dry-run mode for example runs
515- isExampleRun := strings .Contains (opts .splunkURL , "splunk.example.com" )
516-
517- client := resty .New ().
518- SetBaseURL (opts .splunkURL ).
519- SetAuthScheme ("Splunk" ).
520- SetAuthToken (opts .splunkToken ).
521- SetHeader ("Content-Type" , "application/json" ).
522- SetLogger (ZerologRestyLogger {})
523-
524- log .Debug ().Str ("report id" , report .ID ).Int ("results" , len (results )).Msg ("Sending aggregated data to Splunk" )
525-
526- const (
527- resultsBatchSize = 10
528- splunkSizeLimitBytes = 100_000_000 // 100MB. Actual limit is over 800MB, but that's excessive
529- exampleSplunkReportFileName = "example_results/example_splunk_report.json"
530- exampleSplunkResultsFileName = "example_results/example_splunk_results_batch_%d.json"
531- )
532-
533- var (
534- splunkErrs = []error {}
535- resultsBatch = []SplunkTestResult {}
536- successfulResultsSent = 0
537- batchNum = 1
538- )
539-
540- for resultCount , result := range results {
541- // No need to send log outputs to Splunk
542- result .FailedOutputs = nil
543- result .PassedOutputs = nil
544- result .PackageOutputs = nil
545-
546- resultsBatch = append (resultsBatch , SplunkTestResult {
547- Event : SplunkTestResultEvent {
548- Event : opts .splunkEvent ,
549- Type : Result ,
550- Data : result ,
551- },
552- SourceType : SplunkSourceType ,
553- Index : SplunkIndex ,
554- })
555-
556- if len (resultsBatch ) >= resultsBatchSize ||
557- resultCount == len (results )- 1 ||
558- binary .Size (resultsBatch ) >= splunkSizeLimitBytes {
559-
560- batchData , testNames , err := batchSplunkResults (resultsBatch )
561- if err != nil {
562- return fmt .Errorf ("error batching results: %w" , err )
563- }
564-
565- if isExampleRun {
566- exampleSplunkResultsFileName := fmt .Sprintf (exampleSplunkResultsFileName , batchNum )
567- exampleSplunkResultsFile , err := os .Create (exampleSplunkResultsFileName )
568- if err != nil {
569- return fmt .Errorf ("error creating example Splunk results file: %w" , err )
570- }
571- for _ , result := range resultsBatch {
572- jsonResult , err := json .Marshal (result )
573- if err != nil {
574- return fmt .Errorf ("error marshaling result for '%s' to json: %w" , result .Event .Data .TestName , err )
575- }
576- _ , err = exampleSplunkResultsFile .Write (jsonResult )
577- if err != nil {
578- return fmt .Errorf ("error writing result for '%s' to file: %w" , result .Event .Data .TestName , err )
579- }
580- }
581- err = exampleSplunkResultsFile .Close ()
582- if err != nil {
583- return fmt .Errorf ("error closing example Splunk results file: %w" , err )
584- }
585- } else {
586- resp , err := client .R ().SetBody (batchData .String ()).Post ("" )
587- if err != nil {
588- splunkErrs = append (splunkErrs ,
589- fmt .Errorf ("error sending results for [%s] to Splunk: %w" , strings .Join (testNames , ", " ), err ),
590- )
591- }
592- if resp .IsError () {
593- splunkErrs = append (splunkErrs ,
594- fmt .Errorf ("error sending result for [%s] to Splunk: %s" , strings .Join (testNames , ", " ), resp .String ()),
595- )
596- }
597- if err == nil && ! resp .IsError () {
598- successfulResultsSent += len (resultsBatch )
599- }
600- }
601- resultsBatch = []SplunkTestResult {}
602- batchNum ++
603- }
604- }
605-
606- if isExampleRun {
607- log .Info ().Msg ("Example Run. See 'example_results/splunk_results' for the results that would be sent to splunk" )
505+ var eg errgroup.Group
506+ eg .Go (func () error {
507+ return sendDataToSplunk (opts , * fullReport )
508+ })
509+ eg .Go (func () error {
510+ return sendDataToDX (opts , * fullReport )
511+ })
512+ if err := eg .Wait (); err != nil {
513+ log .Error ().Err (err ).Msg ("Error sending data to 3rd party services" )
608514 }
609515
610- reportData := SplunkTestReport {
611- Event : SplunkTestReportEvent {
612- Event : opts .splunkEvent ,
613- Type : Report ,
614- Data : report ,
615- Incomplete : len (splunkErrs ) > 0 ,
616- },
617- SourceType : SplunkSourceType ,
618- Index : SplunkIndex ,
619- }
620-
621- if isExampleRun {
622- exampleSplunkReportFile , err := os .Create (exampleSplunkReportFileName )
623- if err != nil {
624- return fmt .Errorf ("error creating example Splunk report file: %w" , err )
625- }
626- jsonReport , err := json .Marshal (reportData )
627- if err != nil {
628- return fmt .Errorf ("error marshaling report: %w" , err )
629- }
630- _ , err = exampleSplunkReportFile .Write (jsonReport )
631- if err != nil {
632- return fmt .Errorf ("error writing report: %w" , err )
633- }
634- log .Info ().Msgf ("Example Run. See '%s' for the results that would be sent to splunk" , exampleSplunkReportFileName )
635- } else {
636- resp , err := client .R ().SetBody (reportData ).Post ("" )
637- if err != nil {
638- splunkErrs = append (splunkErrs , fmt .Errorf ("error sending report '%s' to Splunk: %w" , report .ID , err ))
639- }
640- if resp .IsError () {
641- splunkErrs = append (splunkErrs , fmt .Errorf ("error sending report '%s' to Splunk: %s" , report .ID , resp .String ()))
642- }
643- }
644-
645- if len (splunkErrs ) > 0 {
646- log .Error ().
647- Int ("successfully sent" , successfulResultsSent ).
648- Int ("total results" , len (results )).
649- Errs ("errors" , splunkErrs ).
650- Str ("report id" , report .ID ).
651- Str ("duration" , time .Since (start ).String ()).
652- Msg ("Errors occurred while sending test results to Splunk" )
653- } else {
654- log .Debug ().
655- Int ("successfully sent" , successfulResultsSent ).
656- Int ("total results" , len (results )).
657- Int ("result batches" , batchNum ).
658- Str ("duration" , time .Since (start ).String ()).
659- Str ("report id" , report .ID ).
660- Msg ("All results sent successfully to Splunk" )
661- }
662-
663- return errors .Join (splunkErrs ... )
664- }
665-
666- // batchSplunkResults creates a batch of TestResult objects as individual JSON objects
667- // Splunk doesn't accept JSON arrays, they want individual events as single JSON objects
668- // https://docs.splunk.com/Documentation/Splunk/9.4.0/Data/FormateventsforHTTPEventCollector
669- func batchSplunkResults (results []SplunkTestResult ) (batchData bytes.Buffer , resultTestNames []string , err error ) {
670- for _ , result := range results {
671- data , err := json .Marshal (result )
672- if err != nil {
673- return batchData , nil , fmt .Errorf ("error marshaling result for '%s': %w" , result .Event .Data .TestName , err )
674- }
675- if _ , err := batchData .Write (data ); err != nil {
676- return batchData , nil , fmt .Errorf ("error writing result for '%s': %w" , result .Event .Data .TestName , err )
677- }
678- if _ , err := batchData .WriteRune ('\n' ); err != nil {
679- return batchData , nil , fmt .Errorf ("error writing newline for '%s': %w" , result .Event .Data .TestName , err )
680- }
681- resultTestNames = append (resultTestNames , result .Event .Data .TestName )
682- }
683- return batchData , resultTestNames , nil
684- }
685-
686- // unBatchSplunkResults un-batches a batch of TestResult objects into a slice of TestResult objects
687- func unBatchSplunkResults (batch []byte ) ([]* SplunkTestResult , error ) {
688- results := make ([]* SplunkTestResult , 0 , bytes .Count (batch , []byte {'\n' }))
689- scanner := bufio .NewScanner (bytes .NewReader (batch ))
690-
691- maxCapacity := 1024 * 1024 // 1 MB
692- buf := make ([]byte , maxCapacity )
693- scanner .Buffer (buf , maxCapacity )
694-
695- var pool sync.Pool
696- pool .New = func () interface {} { return new (SplunkTestResult ) }
697-
698- for scanner .Scan () {
699- line := scanner .Bytes ()
700- if len (bytes .TrimSpace (line )) == 0 {
701- continue // Skip empty lines
702- }
703-
704- result := pool .Get ().(* SplunkTestResult )
705- if err := json .Unmarshal (line , result ); err != nil {
706- return results , fmt .Errorf ("error unmarshaling result: %w" , err )
707- }
708- results = append (results , result )
709- }
710-
711- if err := scanner .Err (); err != nil {
712- return results , fmt .Errorf ("error scanning: %w" , err )
713- }
714-
715- return results , nil
516+ return fullReport , err
716517}
717518
718519// aggregateReports aggregates multiple TestReport objects into a single TestReport
0 commit comments