Add Apache Spark receiver (#5318)

* Add apachespark receiver

* Add more docker contents

* Update CHANGELOG.md

* Fix formatting

* Fix golden file test format

* rename test file

* Ignore scope version in testing

* Update internal/components/components.go

* Fix fmt error

Author: crobert-1

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - (Splunk) Update Python to 3.12.5 in the Smart Agent bundle for Linux and Windows. Check [What’s New In Python 3.12](https://docs.python.org/3/whatsnew/3.12.html) for details. ([#5298](https://github.com/signalfx/splunk-otel-collector/pull/5298))
 
+### 🚀 New components 🚀
+
+- (Splunk) Add `apachespark` receiver ([#5318](https://github.com/signalfx/splunk-otel-collector/pull/5318))
+
 ## v0.108.1
 
 This Splunk OpenTelemetry Collector release includes changes from the [opentelemetry-collector v0.108.1](https://github.com/open-telemetry/opentelemetry-collector/releases/tag/v0.108.1) and the [opentelemetry-collector-contrib v0.108.0](https://github.com/open-telemetry/opentelemetry-collector-contrib/releases/tag/v0.108.0) releases where appropriate.

docker/apachespark/Dockerfile

@@ -0,0 +1,8 @@
+FROM apache/spark:3.4.0-python3
+
+COPY ./long_running.py /opt/spark/examples/src/main/python/long_running.py
+RUN chmod +x /opt/spark/examples/src/main/python/long_running.py
+
+CMD [ "/opt/spark/bin/spark-submit", "/opt/spark/examples/src/main/python/long_running.py" ]
+
+EXPOSE 4040

docker/apachespark/long_running.py

@@ -0,0 +1,82 @@
+# Keep the original spark license.
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF 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.
+#
+# Original source: https://github.com/apache/spark/blob/master/examples/src/main/python/status_api_demo.py
+#
+# NOTICE:
+# Apache Spark
+# Copyright 2014 and onwards The Apache Software Foundation.
+#
+# This product includes software developed at
+# The Apache Software Foundation (http://www.apache.org/).
+#
+
+import time
+import threading
+import queue as Queue
+from typing import Any, Callable, List, Tuple
+
+from pyspark import SparkConf, SparkContext
+
+
+def delayed(seconds: int) -> Callable[[Any], Any]:
+    def f(x: int) -> int:
+        time.sleep(seconds)
+        return x
+    return f
+
+
+def call_in_background(f: Callable[..., Any], *args: Any) -> Queue.Queue:
+    result: Queue.Queue = Queue.Queue(1)
+    t = threading.Thread(target=lambda: result.put(f(*args)))
+    t.daemon = True
+    t.start()
+    return result
+
+
+def main() -> None:
+    conf = SparkConf().set("spark.ui.showConsoleProgress", "false")
+    sc = SparkContext(appName="PythonStatusAPIDemo", conf=conf)
+
+    def run() -> List[Tuple[int, int]]:
+        rdd = sc.parallelize(range(10), 10).map(delayed(2))
+        reduced = rdd.map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y)
+        return reduced.map(delayed(2)).collect()
+
+    result = call_in_background(run)
+    status = sc.statusTracker()
+    while result.empty():
+        ids = status.getJobIdsForGroup()
+        for id in ids:
+            job = status.getJobInfo(id)
+            assert job is not None
+
+            print("Job", id, "status: ", job.status)
+            for sid in job.stageIds:
+                info = status.getStageInfo(sid)
+                if info:
+                    print("Stage %d: %d tasks total (%d active, %d complete)" %
+                          (sid, info.numTasks, info.numActiveTasks, info.numCompletedTasks))
+        time.sleep(1)
+
+    print("Job results are:", result.get())
+    sc.stop()
+
+
+if __name__ == "__main__":
+    for i in range(100):
+        main()

docker/docker-compose.yml

@@ -15,6 +15,13 @@ services:
     build: ./apache
     ports:
       - "18080:80"
+  apachespark:
+    image: quay.io/splunko11ytest/apachespark:latest
+    profiles:
+      - integration
+    build: ./apachespark
+    ports:
+      - "4040:4040"
   # Cassandra image for collectd-cassandra test:
   cassandra:
     image: quay.io/splunko11ytest/cassandra:latest

docs/components.md

@@ -15,6 +15,7 @@ The distribution offers support for the following components.
 |:-------------------------------------------------------------------------------------------------------------------------------------------------------------------|:-----------------|
 | [active_directory_ds](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/activedirectorydsreceiver)                              | [beta]           |
 | [apache](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/apachereceiver)                                                      | [alpha]          |
+| [apachespark](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/apachesparkreceiver)                                            | [alpha]          |
 | [awscontainerinsights](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/awscontainerinsightreceiver)                           | [beta]           |
 | [awsecscontainermetrics](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/awsecscontainermetricsreceiver)                      | [beta]           |
 | [azureblob](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/azureblobreceiver)                                                | [alpha]          |

go.mod

@@ -62,6 +62,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.108.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirectorydsreceiver v0.108.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachereceiver v0.108.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachesparkreceiver v0.108.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver v0.108.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsecscontainermetricsreceiver v0.108.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/azureblobreceiver v0.108.0

go.sum

@@ -1394,6 +1394,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirecto
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirectorydsreceiver v0.108.0/go.mod h1:Zn5FJSJ2h6DUw84ZqwPiJFCXZWsEeBo++SV3xNeSuO4=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachereceiver v0.108.0 h1:RFB7p/HUt46C4HDZjwwDDYYv+ipn/SVTpbuLgXzEbZQ=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachereceiver v0.108.0/go.mod h1:7o6+Ojnd9enSTTVUfzORsgMT83qy0Ds0rNl6CXJb5cQ=
+github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachesparkreceiver v0.108.0 h1:alseMqI1EDNxMBxfBMQde3r0g/KUauEWbm/u6vKzjnM=
+github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachesparkreceiver v0.108.0/go.mod h1:GgM4G2KU5/CYVKUpnJkTsud8B6OvOlnNm6gz3KjsuwU=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver v0.108.0 h1:GQPIFe/91bnSymg3/4cfW6kbQD6+VMCX8yPzgaVGoaQ=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver v0.108.0/go.mod h1:BK/JiWZQtGM94UOwa+JWgI6ABUtJExPvM82uTQPXOns=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsecscontainermetricsreceiver v0.108.0 h1:4/SyCI9608uBuk0qkl7BWboXxWlArSys/qHBU/Wk4fQ=

internal/components/components.go

@@ -57,6 +57,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/activedirectorydsreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachereceiver"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/apachesparkreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsecscontainermetricsreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/azureblobreceiver"
@@ -164,6 +165,7 @@ func Get() (otelcol.Factories, error) {
 	receivers, err := receiver.MakeFactoryMap(
 		activedirectorydsreceiver.NewFactory(),
 		apachereceiver.NewFactory(),
+		apachesparkreceiver.NewFactory(),
 		awscontainerinsightreceiver.NewFactory(),
 		awsecscontainermetricsreceiver.NewFactory(),
 		azureblobreceiver.NewFactory(),

internal/components/components_test.go

@@ -46,6 +46,7 @@ func TestDefaultComponents(t *testing.T) {
 	expectedReceivers := []string{
 		"active_directory_ds",
 		"apache",
+		"apachespark",
 		"awscontainerinsightreceiver",
 		"awsecscontainermetrics",
 		"azureblob",

tests/receivers/apachespark/apachespark_test.go

@@ -0,0 +1,43 @@
+// Copyright Splunk, Inc.
+//
+// Licensed 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.
+
+//go:build integration
+
+package tests
+
+import (
+	"testing"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
+
+	"github.com/signalfx/splunk-otel-collector/tests/testutils"
+)
+
+const sparkPort = "4040"
+
+func TestApacheSparkIntegration(t *testing.T) {
+	testutils.CheckGoldenFile(t, "all_metrics_config.yaml", "all_expected_metrics.yaml",
+		pmetrictest.IgnoreMetricValues(),
+		pmetrictest.IgnoreStartTimestamp(),
+		pmetrictest.IgnoreTimestamp(),
+		pmetrictest.IgnoreScopeVersion(),
+		pmetrictest.IgnoreResourceAttributeValue("spark.application.id"),
+		pmetrictest.IgnoreResourceAttributeValue("spark.application.name"),
+		pmetrictest.IgnoreMetricAttributeValue("active", "spark.stage.status"),
+		pmetrictest.IgnoreMetricAttributeValue("complete", "spark.stage.status"),
+		pmetrictest.IgnoreMetricAttributeValue("failed", "spark.stage.status"),
+		pmetrictest.IgnoreMetricAttributeValue("pending", "spark.stage.status"),
+		pmetrictest.IgnoreMetricDataPointsOrder(),
+	)
+}