support: top-trace with some file exporters

builder-config.yml

@@ -6,6 +6,7 @@ exporters:
   - gomod: go.opentelemetry.io/collector/exporter/debugexporter v0.102.0
   - gomod: go.opentelemetry.io/collector/exporter/otlpexporter v0.102.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/elasticsearchexporter v0.102.0
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/fileexporter v0.102.0
 
 processors:
   - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.102.0

config.yml

@@ -18,14 +18,16 @@ extensions:
 exporters:
   debug:
     verbosity: basic
+  file:
+    path: /tmp/logs.txt
 
 service:
   extensions: [zpages, pprof]
   pipelines:
     traces:
       receivers: [githubactionsjunit]
       processors: [batch]
-      exporters: [debug]
+      exporters: [debug, file]
   telemetry:
     logs:
       level: debug

receiver.go

@@ -17,8 +17,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
-	"go.opentelemetry.io/otel/attribute"
-	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
 	"go.uber.org/zap"
 )
 
@@ -30,19 +28,21 @@ func newTracesReceiver(cfg *Config, params receiver.CreateSettings, nextConsumer
 	}
 	params.Logger.Info("GitHub API rate limit", zap.Int("limit", rateLimit.GetCore().Limit), zap.Int("remaining", rateLimit.GetCore().Remaining), zap.Time("reset", rateLimit.GetCore().Reset.Time))
 	return &githubactionsjunitReceiver{
-		config:   cfg,
-		settings: params,
-		logger:   params.Logger,
-		ghClient: ghClient,
+		nextConsumer: nextConsumer,
+		config:       cfg,
+		settings:     params,
+		logger:       params.Logger,
+		ghClient:     ghClient,
 	}, nil
 }
 
 type githubactionsjunitReceiver struct {
-	config   *Config
-	server   *http.Server
-	settings receiver.CreateSettings
-	logger   *zap.Logger
-	ghClient *github.Client
+	nextConsumer consumer.Traces
+	config       *Config
+	server       *http.Server
+	settings     receiver.CreateSettings
+	logger       *zap.Logger
+	ghClient     *github.Client
 }
 
 func (rec *githubactionsjunitReceiver) Start(ctx context.Context, host component.Host) error {
@@ -122,7 +122,7 @@ func (rec *githubactionsjunitReceiver) handleWorkflowRunEvent(workflowRunEvent *
 		return
 	}
 	for _, artifact := range junitArtifacts {
-		err := processArtifact(rec.logger, rec.ghClient, workflowRunEvent, artifact)
+		err := processArtifact(rec.logger, rec.ghClient, rec.config, workflowRunEvent, artifact, r.Context(), rec.nextConsumer)
 		if err != nil {
 			// TODO: report error but keep processing other artifacts
 			rec.logger.Error("Failed to process artifact", zap.Error(err))
@@ -133,7 +133,7 @@ func (rec *githubactionsjunitReceiver) handleWorkflowRunEvent(workflowRunEvent *
 	}
 }
 
-func processArtifact(logger *zap.Logger, ghClient *github.Client, workflowRunEvent *github.WorkflowRunEvent, artifact *github.Artifact) error {
+func processArtifact(logger *zap.Logger, ghClient *github.Client, config *Config, workflowRunEvent *github.WorkflowRunEvent, artifact *github.Artifact, ctx context.Context, nextConsumer consumer.Traces) error {
 	zipFile, err := downloadArtifact(context.Background(), ghClient, workflowRunEvent, artifact)
 	if err != nil {
 		return err
@@ -142,8 +142,26 @@ func processArtifact(logger *zap.Logger, ghClient *github.Client, workflowRunEve
 	for _, file := range zipFile.Reader.File {
 		logger.Debug("Processing file", zap.String("artifact", artifact.GetName()), zap.String("file", file.Name))
 		suites := processJunitFile(file, logger)
-		for _, suite := range suites {
-			processSuite(suite, logger)
+
+		// Convert the Test Suites to OpenTelemetry traces
+		td, err := suitesToTraces(suites, workflowRunEvent, config, logger)
+		if err != nil {
+			logger.Debug("Failed to convert event to traces", zap.Error(err))
+			// Move forward with the next file
+		}
+
+		// Check if the traces data contains any ResourceSpans
+		if td.ResourceSpans().Len() > 0 {
+			spanCount := td.SpanCount()
+			logger.Debug("Unmarshaled spans", zap.Int("#spans", spanCount))
+			td.ResourceSpans()
+
+			consumerErr := nextConsumer.ConsumeTraces(ctx, td)
+			if consumerErr != nil {
+				logger.Debug("Failed to process traces", zap.Error(consumerErr))
+			}
+		} else {
+			logger.Debug("No spans to unmarshal or traces not initialized")
 		}
 	}
 	return nil
@@ -175,55 +193,6 @@ func processJunitFile(file *zip.File, logger *zap.Logger) []junit.Suite {
 	return suites
 }
 
-func processSuite(suite junit.Suite, logger *zap.Logger) {
-
-	// Set up the attributes for the suite
-	suiteAttributes := []attribute.KeyValue{
-		semconv.CodeNamespaceKey.String(suite.Package),
-		attribute.Key(TestsSuiteName).String(suite.Name),
-		attribute.Key(TestsSystemErr).String(suite.SystemErr),
-		attribute.Key(TestsSystemOut).String(suite.SystemOut),
-		attribute.Key(TestsDuration).Int64(suite.Totals.Duration.Milliseconds()),
-	}
-
-	// Add suite properties as labels
-	suiteAttributes = append(suiteAttributes, propsToLabels(suite.Properties)...)
-
-	// For each test in the suite, set up the attributes
-	for _, test := range suite.Tests {
-		testAttributes := []attribute.KeyValue{
-			semconv.CodeFunctionKey.String(test.Name),
-			attribute.Key(TestDuration).Int64(test.Duration.Milliseconds()),
-			attribute.Key(TestClassName).String(test.Classname),
-			attribute.Key(TestMessage).String(test.Message),
-			attribute.Key(TestStatus).String(string(test.Status)),
-			attribute.Key(TestSystemErr).String(test.SystemErr),
-			attribute.Key(TestSystemOut).String(test.SystemOut),
-		}
-
-		testAttributes = append(testAttributes, propsToLabels(test.Properties)...)
-		testAttributes = append(testAttributes, suiteAttributes...)
-
-		if test.Error != nil {
-			testAttributes = append(testAttributes, attribute.Key(TestError).String(test.Error.Error()))
-		}
-		var stringSlice []string
-		for _, attr := range testAttributes {
-			stringSlice = append(stringSlice, fmt.Sprintf("%s: %v", attr.Key, attr.Value.AsString()))
-		}
-		logger.Debug("Processing test suite", zap.Strings("attributes", stringSlice))
-	}
-}
-
-func propsToLabels(props map[string]string) []attribute.KeyValue {
-	attributes := []attribute.KeyValue{}
-	for k, v := range props {
-		attributes = append(attributes, attribute.Key(k).String(v))
-	}
-
-	return attributes
-}
-
 func getArtifacts(ctx context.Context, ghEvent *github.WorkflowRunEvent, ghClient *github.Client) ([]*github.Artifact, error) {
 	listArtifactsOpts := &github.ListOptions{
 		PerPage: 100,

trace_attributes.go

@@ -0,0 +1,101 @@
+package opentelemetrygithubactionsjunitreceiver
+
+import (
+	"strings"
+	"time"
+
+	"github.com/google/go-github/v62/github"
+	"github.com/joshdk/go-junit"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
+)
+
+func createResourceAttributesTestSuite(resource pcommon.Resource, suite junit.Suite, config *Config) {
+	attrs := resource.Attributes()
+	attrs.PutStr(string(semconv.CodeNamespaceKey), suite.Package)
+	attrs.PutStr("tests.suite.suitename", suite.Name)
+	attrs.PutStr("tests.suite.suitename", suite.Name)
+	attrs.PutStr("tests.suite.systemerr", suite.SystemErr)
+	attrs.PutStr("tests.suite.systemout", suite.SystemOut)
+	attrs.PutInt("tests.suite.duration", suite.Totals.Duration.Milliseconds())
+}
+
+func createResourceAttributesTest(resource pcommon.Resource, test junit.Test, config *Config) {
+	attrs := resource.Attributes()
+	attrs.PutInt("tests.case.duration", test.Duration.Milliseconds())
+	attrs.PutStr(string(semconv.CodeFunctionKey), test.Name)
+	attrs.PutStr("tests.case.classname", test.Classname)
+	attrs.PutStr("tests.case.message", test.Message)
+	attrs.PutStr("tests.case.status", string(test.Status))
+	attrs.PutStr("tests.case.systemerr", test.SystemErr)
+	attrs.PutStr("tests.case.systemout", test.SystemOut)
+	if test.Error != nil {
+		attrs.PutStr("tests.case.error", test.Error.Error())
+	}
+}
+
+func createResourceAttributes(resource pcommon.Resource, e *github.WorkflowRunEvent, config *Config) {
+	attrs := resource.Attributes()
+
+	serviceName := generateServiceName(config, e.GetRepo().GetFullName())
+	attrs.PutStr("service.name", serviceName)
+
+	attrs.PutStr("ci.github.workflow.run.actor.login", e.GetWorkflowRun().GetActor().GetLogin())
+
+	attrs.PutStr("ci.github.workflow.run.conclusion", e.GetWorkflowRun().GetConclusion())
+	attrs.PutStr("ci.github.workflow.run.created_at", e.GetWorkflowRun().GetCreatedAt().Format(time.RFC3339))
+	attrs.PutStr("ci.github.workflow.run.display_title", e.GetWorkflowRun().GetDisplayTitle())
+	attrs.PutStr("ci.github.workflow.run.event", e.GetWorkflowRun().GetEvent())
+	attrs.PutStr("ci.github.workflow.run.head_branch", e.GetWorkflowRun().GetHeadBranch())
+	attrs.PutStr("ci.github.workflow.run.head_sha", e.GetWorkflowRun().GetHeadSHA())
+	attrs.PutStr("ci.github.workflow.run.html_url", e.GetWorkflowRun().GetHTMLURL())
+	attrs.PutInt("ci.github.workflow.run.id", e.GetWorkflowRun().GetID())
+	attrs.PutStr("ci.github.workflow.run.name", e.GetWorkflowRun().GetName())
+	attrs.PutStr("ci.github.workflow.run.path", e.GetWorkflow().GetPath())
+
+	if e.GetWorkflowRun().GetPreviousAttemptURL() != "" {
+		htmlURL := transformGitHubAPIURL(e.GetWorkflowRun().GetPreviousAttemptURL())
+		attrs.PutStr("ci.github.workflow.run.previous_attempt_url", htmlURL)
+	}
+
+	if len(e.GetWorkflowRun().ReferencedWorkflows) > 0 {
+		var referencedWorkflows []string
+		for _, workflow := range e.GetWorkflowRun().ReferencedWorkflows {
+			referencedWorkflows = append(referencedWorkflows, workflow.GetPath())
+		}
+		attrs.PutStr("ci.github.workflow.run.referenced_workflows", strings.Join(referencedWorkflows, ";"))
+	}
+
+	attrs.PutInt("ci.github.workflow.run.run_attempt", int64(e.GetWorkflowRun().GetRunAttempt()))
+	attrs.PutStr("ci.github.workflow.run.run_started_at", e.GetWorkflowRun().RunStartedAt.Format(time.RFC3339))
+	attrs.PutStr("ci.github.workflow.run.status", e.GetWorkflowRun().GetStatus())
+	attrs.PutStr("ci.github.workflow.run.sender.login", e.GetSender().GetLogin())
+	attrs.PutStr("ci.github.workflow.run.triggering_actor.login", e.GetWorkflowRun().GetTriggeringActor().GetLogin())
+	attrs.PutStr("ci.github.workflow.run.updated_at", e.GetWorkflowRun().GetUpdatedAt().Format(time.RFC3339))
+	duration := e.GetWorkflowRun().GetUpdatedAt().Sub(e.GetWorkflowRun().GetRunStartedAt().Time)
+	attrs.PutInt("ci.github.workflow.run.duration_millis", duration.Milliseconds())
+
+	attrs.PutStr("ci.system", "github")
+
+	attrs.PutStr("scm.system", "git")
+
+	attrs.PutStr("scm.git.head_branch", e.GetWorkflowRun().GetHeadBranch())
+	attrs.PutStr("scm.git.head_commit.author.email", e.GetWorkflowRun().GetHeadCommit().GetAuthor().GetEmail())
+	attrs.PutStr("scm.git.head_commit.author.name", e.GetWorkflowRun().GetHeadCommit().GetAuthor().GetName())
+	attrs.PutStr("scm.git.head_commit.committer.email", e.GetWorkflowRun().GetHeadCommit().GetCommitter().GetEmail())
+	attrs.PutStr("scm.git.head_commit.committer.name", e.GetWorkflowRun().GetHeadCommit().GetCommitter().GetName())
+	attrs.PutStr("scm.git.head_commit.message", e.GetWorkflowRun().GetHeadCommit().GetMessage())
+	attrs.PutStr("scm.git.head_commit.timestamp", e.GetWorkflowRun().GetHeadCommit().GetTimestamp().Format(time.RFC3339))
+	attrs.PutStr("scm.git.head_sha", e.GetWorkflowRun().GetHeadSHA())
+
+	if len(e.GetWorkflowRun().PullRequests) > 0 {
+		var prUrls []string
+		for _, pr := range e.GetWorkflowRun().PullRequests {
+			prUrls = append(prUrls, convertPRURL(pr.GetURL()))
+		}
+		attrs.PutStr("scm.git.pull_requests.url", strings.Join(prUrls, ";"))
+	}
+
+	attrs.PutStr("scm.git.repo", e.GetRepo().GetFullName())
+
+}