The Crawler now registers itself at the instance registry #24
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| io.swagger.client { | ||
|
|
||
| apiRequest { | ||
|
|
||
| compression { | ||
| enabled: false | ||
| size-threshold: 0 | ||
| } | ||
|
|
||
| trust-certificates: true | ||
|
|
||
| connection-timeout: 5000ms | ||
|
|
||
| default-headers { | ||
| "userAgent": "swagger-client_1.0.0" | ||
| } | ||
|
|
||
| // let you define custom http status code, as in : | ||
| // { code: 601, reason: "some custom http status code", success: false } | ||
| custom-codes : [] | ||
| } | ||
| } | ||
|
|
||
| spray.can.host-connector.max-redirects = 10 |
This file contains hidden or 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 hidden or 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 hidden or 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 |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| // Copyright (C) 2018 The Delphi Team. | ||
| // See the LICENCE file distributed with this work for additional | ||
| // information regarding copyright ownership. | ||
| // | ||
| // 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. | ||
|
|
||
| package de.upb.cs.swt.delphi.crawler.instancemanagement | ||
|
|
||
| import java.net.InetAddress | ||
|
|
||
| import akka.actor.ActorSystem | ||
| import akka.http.scaladsl.Http | ||
| import akka.http.scaladsl.marshalling.Marshal | ||
| import akka.http.scaladsl.model._ | ||
| import akka.http.scaladsl.unmarshalling.Unmarshal | ||
| import akka.stream.ActorMaterializer | ||
| import de.upb.cs.swt.delphi.crawler.{AppLogging, Configuration, Crawler} | ||
| import de.upb.cs.swt.delphi.crawler.io.swagger.client.model.InstanceEnums.ComponentType | ||
| import de.upb.cs.swt.delphi.crawler.io.swagger.client.model.{Instance, JsonSupport} | ||
|
|
||
| import scala.concurrent.duration.Duration | ||
| import scala.concurrent.{Await, ExecutionContext, Future} | ||
| import scala.util.{Failure, Success, Try} | ||
|
|
||
| object InstanceRegistry extends JsonSupport with AppLogging | ||
| { | ||
|
|
||
| implicit val system : ActorSystem = Crawler.system | ||
| implicit val ec : ExecutionContext = system.dispatcher | ||
| implicit val materializer : ActorMaterializer = Crawler.materializer | ||
|
|
||
|
|
||
| def register(configuration: Configuration) : Try[Long] = { | ||
| val instance = createInstance(None,configuration.controlServerPort, configuration.instanceName) | ||
|
|
||
| Await.result(postInstance(instance, configuration.instanceRegistryUri + "/register") map {response => | ||
| if(response.status == StatusCodes.OK){ | ||
| Await.result(Unmarshal(response.entity).to[String] map { assignedID => | ||
| val id = assignedID.toLong | ||
| log.info(s"Successfully registered at Instance Registry, got ID $id.") | ||
| Success(id) | ||
| } recover { case ex => | ||
| log.warning(s"Failed to read assigned ID from Instance Registry, exception: $ex") | ||
| Failure(ex) | ||
| }, Duration.Inf) | ||
| } | ||
| else { | ||
| val statuscode = response.status | ||
| log.warning(s"Failed to register at Instance Registry, server returned $statuscode") | ||
| Failure(new RuntimeException(s"Failed to register at Instance Registry, server returned $statuscode")) | ||
| } | ||
|
|
||
| } recover {case ex => | ||
| log.warning(s"Failed to register at Instance Registry, exception: $ex") | ||
| Failure(ex) | ||
| }, Duration.Inf) | ||
| } | ||
|
|
||
| def retrieveElasticSearchInstance(configuration: Configuration) : Try[Instance] = { | ||
| if(!configuration.usingInstanceRegistry) { | ||
| Failure(new RuntimeException("Cannot get ElasticSearch instance from Instance Registry, no Instance Registry available.")) | ||
| } else { | ||
| val request = HttpRequest(method = HttpMethods.GET, configuration.instanceRegistryUri + "/matchingInstance?ComponentType=ElasticSearch") | ||
|
|
||
| Await.result(Http(system).singleRequest(request) map {response => | ||
| val status = response.status | ||
| if(status == StatusCodes.OK) { | ||
|
|
||
| Await.result(Unmarshal(response.entity).to[Instance] map {instance => | ||
| val elasticIP = instance.iP | ||
| log.info(s"Instance Registry assigned ElasticSearch instance at ${elasticIP.getOrElse("None")}") | ||
| Success(instance) | ||
| } recover {case ex => | ||
| log.warning(s"Failed to read response from Instance Registry, exception: $ex") | ||
| Failure(ex) | ||
| }, Duration.Inf) | ||
| } | ||
| else{ | ||
| log.warning(s"Failed to read response from Instance Registry, server returned $status") | ||
| Failure(new RuntimeException(s"Failed to read response from Instance Registry, server returned $status")) | ||
| } | ||
| } recover { case ex => | ||
| log.warning(s"Failed to request ElasticSearch instance from Instance Registry, exception: $ex ") | ||
| Failure(ex) | ||
| }, Duration.Inf) | ||
| } | ||
| } | ||
|
|
||
| def sendMatchingResult(isElasticSearchReachable : Boolean, configuration: Configuration) : Try[Unit] = { | ||
| if(!configuration.usingInstanceRegistry) { | ||
| Failure(new RuntimeException("Cannot post matching result to Instance Registry, no Instance Registry available.")) | ||
| } else { | ||
| if(configuration.elasticsearchInstance.iD.isEmpty) { | ||
| Failure(new RuntimeException("Cannot post matching result to Instance Registry, assigned ElasticSearch instance has no ID.")) | ||
| } else { | ||
| val idToPost = configuration.elasticsearchInstance.iD.getOrElse(-1L) | ||
bhermann marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| val request = HttpRequest( | ||
| method = HttpMethods.POST, | ||
| configuration.instanceRegistryUri + s"/matchingResult?Id=$idToPost&MatchingSuccessful=$isElasticSearchReachable") | ||
|
|
||
| Await.result(Http(system).singleRequest(request) map {response => | ||
| if(response.status == StatusCodes.OK){ | ||
| log.info("Successfully posted matching result to Instance Registry.") | ||
| Success() | ||
| } | ||
| else { | ||
| val statuscode = response.status | ||
| log.warning(s"Failed to post matching result to Instance Registry, server returned $statuscode") | ||
| Failure(new RuntimeException(s"Failed to post matching result to Instance Registry, server returned $statuscode")) | ||
| } | ||
|
|
||
| } recover {case ex => | ||
| log.warning(s"Failed to post matching result to Instance Registry, exception: $ex") | ||
| Failure(new RuntimeException(s"Failed to post matching result tot Instance Registry, exception: $ex")) | ||
| }, Duration.Inf) | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| def deregister(configuration: Configuration) : Try[Unit] = { | ||
| if(!configuration.usingInstanceRegistry){ | ||
| Failure(new RuntimeException("Cannot deregister from Instance Registry, no Instance Registry available.")) | ||
| } else { | ||
| val id : Long = configuration.assignedID.getOrElse(-1L) | ||
|
|
||
| val request = HttpRequest(method = HttpMethods.POST, configuration.instanceRegistryUri + s"/deregister?Id=$id") | ||
|
|
||
| Await.result(Http(system).singleRequest(request) map {response => | ||
| if(response.status == StatusCodes.OK){ | ||
| log.info("Successfully deregistered from Instance Registry.") | ||
| Success() | ||
| } | ||
| else { | ||
| val statuscode = response.status | ||
| log.warning(s"Failed to deregister from Instance Registry, server returned $statuscode") | ||
| Failure(new RuntimeException(s"Failed to deregister from Instance Registry, server returned $statuscode")) | ||
| } | ||
|
|
||
| } recover {case ex => | ||
| log.warning(s"Failed to deregister to Instance Registry, exception: $ex") | ||
| Failure(ex) | ||
| }, Duration.Inf) | ||
| } | ||
| } | ||
|
|
||
| def postInstance(instance : Instance, uri: String) () : Future[HttpResponse] = | ||
| Marshal(instance).to[RequestEntity] flatMap { entity => | ||
| val request = HttpRequest(method = HttpMethods.POST, uri = uri, entity = entity) | ||
| Http(system).singleRequest(request) | ||
| } | ||
|
|
||
|
|
||
| private def createInstance(id: Option[Long], controlPort : Int, name : String) : Instance = | ||
| Instance(id, Option(InetAddress.getLocalHost.getHostAddress), Option(controlPort), Option(name), Option(ComponentType.Crawler)) | ||
| } | ||
This file contains hidden or 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 |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| /** | ||
| * NOTE: This class is auto generated by the akka-scala (beta) swagger code generator program. | ||
| * https://github.com/swagger-api/swagger-codegen | ||
| * For any issue or feedback, please open a ticket via https://github.com/swagger-api/swagger-codegen/issues/new | ||
| */ | ||
|
|
||
| package de.upb.cs.swt.delphi.crawler.io.swagger.client.model | ||
|
|
||
| import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport | ||
| import spray.json._ | ||
|
|
||
| trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol { | ||
| implicit val componentTypeFormat = new JsonFormat[InstanceEnums.ComponentType] { | ||
| def write(compType : InstanceEnums.ComponentType) = JsString(compType.toString) | ||
|
|
||
| def read(value: JsValue) = value match { | ||
| case JsString(s) => s match { | ||
| case "Crawler" => InstanceEnums.ComponentType.Crawler | ||
| case "WebApi" => InstanceEnums.ComponentType.WebApi | ||
| case "WebApp" => InstanceEnums.ComponentType.WebApp | ||
| case "DelphiManagement" => InstanceEnums.ComponentType.DelphiManagement | ||
| case "ElasticSearch" => InstanceEnums.ComponentType.ElasticSearch | ||
| case x => throw new RuntimeException(s"Unexpected string value $x for component type.") | ||
| } | ||
| case y => throw new RuntimeException(s"Unexpected type $y while deserializing component type.") | ||
| } | ||
| } | ||
| implicit val instanceFormat = jsonFormat5(Instance) | ||
| } | ||
|
|
||
| final case class Instance ( | ||
bhermann marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| iD: Option[Long], | ||
| iP: Option[String], | ||
| portnumber: Option[Long], | ||
| name: Option[String], | ||
| /* Component Type */ | ||
| componentType: Option[InstanceEnums.ComponentType] | ||
| ) | ||
|
|
||
| object InstanceEnums { | ||
|
|
||
| type ComponentType = ComponentType.Value | ||
| object ComponentType extends Enumeration { | ||
| val Crawler = Value("Crawler") | ||
| val WebApi = Value("WebApi") | ||
| val WebApp = Value("WebApp") | ||
| val DelphiManagement = Value("DelphiManagement") | ||
| val ElasticSearch = Value("ElasticSearch") | ||
| } | ||
|
|
||
| } | ||
|
|
||
This file contains hidden or 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
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do we need this local resolver?