-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
f89cff6
commit 143a0f1
Showing
6 changed files
with
38 additions
and
132 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,4 @@ | ||
| minikube_url = 192.168.64.11 | ||
| minikube_url = 192.168.64.24 | ||
|
|
||
| prestodb = { | ||
| db { | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,128 +1,41 @@ | ||
| package it_tests.utils | ||
|
|
||
| import org.json.JSONObject | ||
| import org.yaml.snakeyaml.Yaml | ||
| import java.io.{File, FileInputStream} | ||
| import java.util | ||
|
|
||
| import com.google.gson.internal.LinkedTreeMap | ||
| import io.kubernetes.client.apis.CustomObjectsApi | ||
| import io.kubernetes.client.util.Config | ||
| import io.kubernetes.client.Configuration | ||
| import io.kubernetes.client.models._ | ||
| import rx.lang.scala.Observable | ||
| import io.circe.generic.auto._ | ||
| import io.circe.parser._ | ||
|
|
||
| import scala.concurrent.{Await, Future} | ||
| import scala.util.{Failure, Success} | ||
| import scala.concurrent.duration._ | ||
| import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext | ||
| import io.fabric8.kubernetes.client.{DefaultKubernetesClient, KubernetesClient} | ||
|
|
||
| class SparkController(crdNamespace: String, resourceName: String) { | ||
| import scala.concurrent.ExecutionContext.Implicits.global | ||
|
|
||
| val client: KubernetesClient = new DefaultKubernetesClient() | ||
| private val crdInstanceName = "spark-pi" | ||
| private val crdGroup = "sparkoperator.k8s.io" | ||
| private val crdVersion = "v1beta1" | ||
| private val crdPlural = "sparkapplications" | ||
| private val crdInstanceName = "spark-pi" | ||
|
|
||
| private val apiClient = Config.defaultClient | ||
| // apiClient.setDebugging(true) | ||
| Configuration.setDefaultApiClient(apiClient) | ||
| private val apiInstance = new CustomObjectsApi | ||
| private val customObjectBody = convertYamlToJson(resourceName) | ||
|
|
||
| private val crdStateStream = Observable | ||
| .interval(Duration(1000, MILLISECONDS)) | ||
| .map(_ => { | ||
| val crdResponse = apiInstance.getNamespacedCustomObject( | ||
| crdGroup, | ||
| crdVersion, | ||
| crdNamespace, | ||
| crdPlural, | ||
| crdInstanceName) | ||
|
|
||
| val JSONResp = new JSONObject(crdResponse.asInstanceOf[LinkedTreeMap[String, Object]]) | ||
| val decoded = decode[CustomObject](JSONResp.toString) | ||
|
|
||
| decoded match { | ||
| case Right(x) => x | ||
| case Left(ex) => throw new RuntimeException(ex) | ||
| } | ||
| }) | ||
| .distinctUntilChanged(x => x.status) | ||
| .share | ||
| private val crdScope = "Namespaced" | ||
| private val crdPodName = "spark-pi-driver" | ||
|
|
||
| private val resource = new FileInputStream(new File(resourceName)) | ||
| val crdContext: CustomResourceDefinitionContext = new CustomResourceDefinitionContext.Builder() | ||
| .withGroup(crdGroup) | ||
| .withName(crdInstanceName) | ||
| .withScope(crdScope) | ||
| .withPlural(crdPlural) | ||
| .withVersion(crdVersion) | ||
| .build() | ||
|
|
||
| def launchSparkTestDeployment(): Unit = { | ||
| val apiCall = | ||
| Future(apiInstance | ||
| .createNamespacedCustomObject( | ||
| crdGroup, | ||
| crdVersion, | ||
| crdNamespace, | ||
| crdPlural, | ||
| customObjectBody.toMap, | ||
| "false")) | ||
| val sparkCustomResource = client.customResource(crdContext) | ||
|
|
||
| apiCall onComplete { | ||
| case Failure(ex) => | ||
| throw new RuntimeException(ex) | ||
| case Success(x) => | ||
| println(s"Startup finished successfully: $x") | ||
| } | ||
| sparkCustomResource.create(crdNamespace, resource) | ||
| Thread.sleep(30000) | ||
|
|
||
| Await.result(apiCall, Duration.Inf) | ||
|
|
||
| crdStateStream | ||
| .takeWhile(x => { | ||
| x.status match { | ||
| case None => true | ||
| case Some(status) => status.applicationState.state != SparkOperatorStatus.RunningState | ||
| } | ||
| }) | ||
| .toBlocking | ||
| .subscribe(x => println(s"Waiting for status ${SparkOperatorStatus.RunningState}: $x")) | ||
|
|
||
| crdStateStream | ||
| .takeWhile(x => { | ||
| x.status match { | ||
| case None => false | ||
| case Some(status) => status.applicationState.state != SparkOperatorStatus.CompletedState | ||
| } | ||
| }) | ||
| .onErrorResumeNext(_ => Observable.empty) | ||
| .subscribe(x => println(s"Monitoring CRD status while running: $x")) | ||
| client.pods() | ||
| .inNamespace(crdNamespace) | ||
| .withName(crdPodName) | ||
| .watchLog(System.out) | ||
| } | ||
|
|
||
| def cleanUpSparkTestDeployment(): Unit = { | ||
| val apiCall = | ||
| Future(apiInstance | ||
| .deleteNamespacedCustomObject( | ||
| crdGroup, | ||
| crdVersion, | ||
| crdNamespace, | ||
| crdPlural, | ||
| crdInstanceName, | ||
| new V1DeleteOptions, | ||
| 0, | ||
| false, | ||
| "Foreground")) | ||
|
|
||
| apiCall onComplete { | ||
| case Failure(ex) => | ||
| println(s"Cleanup failed with error: $ex") | ||
| case Success(x) => | ||
| println(s"Cleanup finished successfully: $x") | ||
| } | ||
|
|
||
| Await.result(apiCall, Duration.Inf) | ||
| } | ||
|
|
||
| private def convertYamlToJson(resourceName: String): JSONObject = { | ||
| val input = new FileInputStream(new File(resourceName)) | ||
| val yaml: Yaml = new Yaml | ||
| val source: util.LinkedHashMap[String, Object] = yaml.load(input) | ||
|
|
||
| new JSONObject(source) | ||
| client.customResource(crdContext).delete(crdNamespace, crdInstanceName) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters