Location Service

The Location Service handles component (i.e., Applications, Sequencers, Assemblies, HCDs, and Services) registration and discovery in the distributed TMT software system.

The CSW Location Service cluster must be running, and appropriate environment variables set to run apps. See CSW Location Server.

A component’s location information can be used by other components and services to connect to it and use it. An example of location information is:

• host address/port pairs
• URL/URIs paths
• connection protocols

Dependencies

To use the Location Service without using the framework, add this to your build.sbt file:

sbt

libraryDependencies += "com.github.tmtsoftware.csw" %% "csw-location-server" % "5.0.1"

Create Location Service

Note that before using this API, the csw-location-server application should be running at a known location in the network (or at multiple locations) and the necessary configuration, environment variables or system properties should be defined to point to the correct host and port number(s).

LocationServiceFactory provides a make method to create an instance of the LocationService API. This call will look for configuration or environment variable settings.

Scala

copysourceprivate val locationService = HttpLocationServiceFactory.makeLocalClient(typedSystem)

Java

copysourceprivate final akka.actor.ActorSystem system = context().system();
private final ILocationService locationService = JHttpLocationServiceFactory.makeLocalClient(Adapter.toTyped(system));

Creating Components, Connections and Registrations

An Application, Sequencer, Assembly, HCD, or Service may need to be used by another component as part of normal observatory operations. It must register its location information with Location Service so that other components can find it. Location information is comprised of:

• ComponentId : A component ID consisting of

  • ComponentName : a name describing the component.
  • ComponentType : such as Container, Sequencer, HCD, Assembly, Service.

• ConnectionType : the means to reach components. These are categorized as Akka, HTTP, or Tcp type connections.

The location information is stored in the Location Service as Registrations.

Some of the examples of the string representation of a connection are:

• TromboneAssembly-assembly-akka
• TromboneHcd-hcd-akka
• ConfigServer-service-http
• EventService-service-tcp

The register API takes a Registration parameter and returns a future registration result. If registration fails for some reason, the returned future will fail with an exception. (Registration will fail if the csw-location-server application is not running or could not be found, or if the given component name was already registered.)

AkkaRegistrationFactory can be used to instantiate AkkaRegistration using AkkaConnection, Actor Ref and optional Metadata. The following example shows registration of both an UnTyped ActorRef and a Typed ActorRef:

Scala

copysource
// add some dummy registrations for illustrative purposes

// dummy http connection
val httpPort = 8080
val httpConnection: HttpConnection = HttpConnection(
  api.models.ComponentId(Prefix(Subsystem.CSW, "configuration"), ComponentType.Service)
)

// When no network type is provided in httpRegistration, default is NetworkType.Private
val httpRegistration: HttpRegistration = HttpRegistration(httpConnection, httpPort, "path123")
private val httpRegResultF: Future[RegistrationResult] = locationService
  .register(httpRegistration)
httpRegResultF
  .map(httpRegResult => {
    log.info(s"$httpRegResult successfully registered in location service")
  })
// When a service wants to register itself on Public network, it can provide NetworkType.Public in httpRegistration
val httpRegistrationOnPublicNetwork: HttpRegistration =
  HttpRegistration(httpConnection, httpPort, "path123", NetworkType.Outside)
private val httpRegResultOnPublicNetworkF: Future[RegistrationResult] =
  locationService.register(httpRegistrationOnPublicNetwork)
httpRegResultOnPublicNetworkF
  .map(httpRegResultOnPublicNetwork => {
    log.info(s"$httpRegResultOnPublicNetwork successfully registered in location service")
  })
// ************************************************************************************************************

// import scaladsl adapter to implicitly convert UnTyped ActorRefs to Typed ActorRef[Nothing]
import akka.actor.typed.scaladsl.adapter.*

// dummy HCD connection
val hcdConnection: AkkaConnection = AkkaConnection(api.models.ComponentId(Prefix(Subsystem.NFIRAOS, "hcd1"), ComponentType.HCD))
val hcdRegistration: AkkaRegistration = AkkaRegistrationFactory.make(
  hcdConnection,
  context
    .actorOf(
      Props(new Actor {
        override def receive: Receive = { case "print" =>
          log.info("hello world")
        }
      }),
      name = "my-actor-1"
    )
    .toTyped
)

// Register UnTyped ActorRef with Location service. Import scaladsl adapter to implicitly convert
// UnTyped ActorRefs to Typed ActorRef[Nothing]
private val hcdRegResultF: Future[RegistrationResult] = locationService
  .register(hcdRegistration)
hcdRegResultF
  .map(hcdRegResult => {
    log.info(s"$hcdRegResult successfully registered in location service")
  })
// ************************************************************************************************************

def behavior(): Behavior[String]    = Behaviors.setup { _ => Behaviors.receiveMessage { _ => Behaviors.same } }
val typedActorRef: ActorRef[String] = context.system.spawn(behavior(), "typed-actor-ref")

val assemblyConnection: AkkaConnection = AkkaConnection(
  ComponentId(Prefix(Subsystem.NFIRAOS, "assembly1"), ComponentType.Assembly)
)

// Register Typed ActorRef[String] with Location Service
val assemblyRegistration: AkkaRegistration = AkkaRegistrationFactory.make(assemblyConnection, typedActorRef)

private val assemblyRegResultF: Future[RegistrationResult] = locationService
  .register(assemblyRegistration)
assemblyRegResultF
  .map(assemblyRegResult => {
    log.info(s"$assemblyRegResult successfully registered in location service")
  })

Java

copysource
// dummy http connection
HttpConnection httpConnection = new HttpConnection(new ComponentId(new Prefix(JSubsystem.CSW,
        "configuration"), JComponentType.Service));

// When no network type is provided in httpRegistration, default is JNetworkType.Private
HttpRegistration httpRegistration = new HttpRegistration(httpConnection, 8080, "path123");
httpRegResult = locationService.register(httpRegistration).get();

// When a service wants to register itself on Public network, it can provide JNetworkType.Public in httpRegistration
HttpRegistration httpRegistrationOnPublicNetwork = new HttpRegistration(httpConnection, 8080, "path123",
        JNetworkType.Outside);
httpRegResultonPublicNetwork = locationService.register(httpRegistrationOnPublicNetwork).get();

// ************************************************************************************************************

// dummy HCD connection
AkkaConnection hcdConnection = new AkkaConnection(new ComponentId(new Prefix(JSubsystem.NFIRAOS, "hcd1"),
        JComponentType.HCD));
ActorRef actorRefUntyped = getContext().actorOf(Props.create(AbstractActor.class, () -> new AbstractActor() {
            @Override
            public Receive createReceive() {
                return ReceiveBuilder.create().build();
            }
        }),
        "my-actor-1"
);

// Register UnTyped ActorRef with Location service. Use javadsl Adapter to convert UnTyped ActorRefs
// to Typed ActorRef[Nothing]

akka.actor.typed.ActorRef actorRef = Adapter.toTyped(actorRefUntyped);
AkkaRegistration hcdRegistration = JAkkaRegistrationFactory.make(hcdConnection, actorRef);
hcdRegResult = locationService.register(hcdRegistration).get();

// ************************************************************************************************************

Behavior<String> behavior = Behaviors.setup(ctx -> Behaviors.same());
akka.actor.typed.ActorRef<String> typedActorRef = Adapter.spawn(context(), behavior, "typed-actor-ref");

AkkaConnection assemblyConnection = new AkkaConnection(new ComponentId(new Prefix(JSubsystem.NFIRAOS,
        "assembly1"), JComponentType.Assembly));

// Register Typed ActorRef[String] with Location Service
AkkaRegistration assemblyRegistration = JAkkaRegistrationFactory.make(assemblyConnection, typedActorRef);


assemblyRegResult = locationService.register(assemblyRegistration).get();

Capability to add metadata while registering connection

Location service supports to add additional information while registering connection. This information is metadata for that registration. Metadata given at point of registration is reflected in location model. This metadata information can be used to store additional info e.g. agentId for any component running on that agent. Metadata field is optional while registration, if not provided location will be reflected with empty metadata

Following example shows adding metadata in HttpRegistration. Similarly, metadata can be added in AkkaRegistration as well as TcpRegistration. Once location is registered Metadata associated can be used for any computation.

Scala

copysource// add some dummy registrations for illustrative purposes

// dummy http connection
val httpPortForService = 8080
val httpConnectionForService: HttpConnection = HttpConnection(
  api.models.ComponentId(Prefix(Subsystem.CSW, "configuration"), ComponentType.Service)
)

// When no network type is provided in httpRegistration, default is NetworkType.Private
val httpRegistrationForService: HttpRegistration =
  HttpRegistration(httpConnectionForService, httpPortForService, "path123", Metadata(Map("key1" -> "value1")))
private val httpRegResultForServiceF: Future[RegistrationResult] = locationService
  .register(httpRegistrationForService)

httpRegResultForServiceF
  .map(httpRegResultForService => {
    log.info(s"$httpRegResultForService successfully registered in location service")
  })

Java

copysource
// dummy http connection
HttpConnection httpConnectionForService = new HttpConnection(new ComponentId(new Prefix(JSubsystem.CSW,
        "configuration"), JComponentType.Service));

// When no network type is provided in httpRegistration, default is JNetworkType.Private
HttpRegistration httpRegistrationForService = new HttpRegistration(httpConnectionForService, 8080, "path123", new Metadata(Map.of("key1", "value1")));
httpRegResult = locationService.register(httpRegistrationForService).get();

// usage of metadata
Map metadataMap = httpRegResult.location().metadata().jValue();

Creating ActorRef for Registration

The ActorSystem used to start actors that will be registered in the Location Service must be created using an ActorSystemFactory as follows:

Scala

copysourceval typedSystem: typed.ActorSystem[SpawnProtocol.Command] =
  ActorSystemFactory.remote(SpawnProtocol(), "csw-examples-locationServiceClient")

Java

copysourceActorSystem<SpawnProtocol.Command> typedSystem = ActorSystemFactory.remote(SpawnProtocol.create(), "csw" +
        "-examples-locationServiceClient");

This is required to start a remote ActorSystem on the same network interface where the csw-cluster is running. All the ActorRefs created using this ActorSystem will be available for communication from other components that are part of the CSW Cluster.

Resolving Connections

The list API returns a list of the currently registered connections from the Location Service.

A connection of interest can be looked up using the resolve or find methods:

resolve gets the location for a connection from the local cache. If not found in the cache, it waits for the event to arrive within the specified time limit and returns None on failure.

find returns the location for a connection from the local cache and returns None if not immediately found there.

Scala

copysource// find connection to LocationServiceExampleComponent in location service
// [do this before starting LocationServiceExampleComponent.  this should return Future[None]]
val exampleConnection: AkkaConnection = LocationServiceExampleComponent.connection

log.info(
  s"Attempting to find $exampleConnection",
  Map(Keys.OBS_ID -> "foo_obs_id", "exampleConnection" -> exampleConnection.name)
)
locationService.find(exampleConnection).map {
  case Some(findResult) =>
    log.info(s"Result of the find call: $findResult")
    val akkaLocation = findResult
    // If the component type is HCD or Assembly, use this to get the correct ActorRef
    val typedComponentRef: ActorRef[ComponentMessage] = akkaLocation.componentRef

    // If the component type is Container, use this to get the correct ActorRef
    val typedContainerRef: ActorRef[ContainerMessage] = akkaLocation.containerRef
  case None => // do something when nothing is found
}

Java

copysource// find connection to LocationServiceExampleComponent in location service
// [do this before starting LocationServiceExampleComponent.  this should return Future[None]]

log.info("Attempting to find " + exampleConnection,
        Map.of(
                JKeys.OBS_ID, "foo_obs_id",
                "exampleConnection", exampleConnection.name()
        ));

Optional<AkkaLocation> findResult = locationService.find(exampleConnection).get();
if (findResult.isPresent()) {
    log.info("Find result: " + connectionInfo(findResult.orElseThrow().connection()));
} else {
    log.info(() -> "Result of the find call : None");
}

The logging output from the above example when the given component is not registered should include:

[INFO] Attempting to find connection AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly))
[INFO] Result of the find call: None 

An example of the resolve command is shown in the following:

Scala

copysource// resolve connection to LocationServiceExampleComponent
// [start LocationServiceExampleComponent after this command but before timeout]
log.info(s"Attempting to resolve $exampleConnection with a wait of $waitForResolveLimit ...")

val resolveResultF: Future[Option[AkkaLocation]] = locationService.resolve(exampleConnection, waitForResolveLimit)
val resolveResult: Option[AkkaLocation]          = Await.result(resolveResultF, waitForResolveLimit + timeout)
resolveResult match {
  case Some(result) =>
    log.info(s"Resolve result: ${locationInfoToString(result)}")
  case None =>
    log.info(s"Timeout waiting for location $exampleConnection to resolve.")
}

Java

copysource// resolve connection to LocationServiceExampleComponent
// [start LocationServiceExampleComponent after this command but before timeout]
Duration waitForResolveLimit = Duration.ofSeconds(30);

log.info(() -> "Attempting to resolve " + exampleConnection + " with a wait of " + waitForResolveLimit + ".." +
        ".", () -> Map.of(
        JKeys.OBS_ID, "foo_obs_id",
        "exampleConnection", exampleConnection.name()
));

Optional<AkkaLocation> resolveResult = locationService.resolve(exampleConnection, waitForResolveLimit).get();
if (resolveResult.isPresent()) {
    log.info(() -> "Resolve result: " + connectionInfo(resolveResult.orElseThrow().connection()));
} else {
    log.info(() -> "Timeout waiting for location " + exampleConnection + " to resolve.");
}

The logging output from the above example should include:

[INFO] Attempting to resolve AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly)) with a wait of 30 seconds ...

If you then start the LocationServiceExampleComponentApp, the following message will be logged:

[INFO] Resolve result: LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType

If not, eventually the operation will timeout and the output should read:

[INFO] Timeout waiting for location AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly)) to resolve.

Filtering

The list API and its variants offer means to inquire about available connections with the Location Service. The parameter-less list returns all available connections

Scala

copysource// list connections in location service
locationService.list.map(connectionList => {
  log.info("All Registered Connections:")
  connectionList.foreach(c => log.info(s"--- ${locationInfoToString(c)}"))
})

Java

copysource// list connections in location service
List<Location> connectionList = locationService.list().get();
log.info("All Registered Connections:");
for (Location loc : connectionList) {
    log.info("--- " + connectionInfo(loc.connection()));
}

The log output from the above should contain:

[INFO] All Registered Connections:
[INFO] --- configuration-service-http, component type=Service, connection type=HttpType
[INFO] --- hcd1-hcd-akka, component type=HCD, connection type=AkkaType
[INFO] --- LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType
[INFO] --- redis-service-tcp, component type=Service, connection type=TcpType
[INFO] --- assembly1-assembly-akka, component type=Assembly, connection type=AkkaType

Other variants are filters using ConnectionType, ComponentType, and hostname.

Filtering by component type is shown below:

Scala

copysource// filter connections based on component type
locationService
  .list(ComponentType.Assembly)
  .map(componentList => {
    log.info("Registered Assemblies:")
    componentList.foreach(c => log.info(s"--- ${locationInfoToString(c)}"))
  })

Java

copysource// filter connections based on component type
List<Location> componentList = locationService.list(JComponentType.Assembly).get();
log.info("Registered Assemblies:");
for (Location loc : componentList) {
    log.info("--- " + connectionInfo(loc.connection()));
}

The log output from the above code should contain:

[INFO] Registered Assemblies:
[INFO] --- LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType
[INFO] --- assembly1-assembly-akka, component type=Assembly, connection type=AkkaType

Filtering by connection type is shown below:

Scala

copysource// filter connections based on connection type
locationService
  .list(ConnectionType.AkkaType)
  .map(akkaList => {
    log.info("Registered Akka connections:")
    akkaList.foreach(c => log.info(s"--- ${locationInfoToString(c)}"))
  })

Java

copysource// filter connections based on connection type
List<Location> akkaList = locationService.list(JConnectionType.AkkaType).get();
log.info("Registered Akka connections:");
for (Location loc : akkaList) {
    log.info("--- " + connectionInfo(loc.connection()));
}

The log output should contain:

[INFO] Registered Akka connections:
[INFO] --- hcd1-hcd-akka, component type=HCD, connection type=AkkaType
[INFO] --- LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType
[INFO] --- assembly1-assembly-akka, component type=Assembly, connection type=AkkaType

Filtering akka connections by prefix is shown below:

Scala

copysource// filter akka locations based on prefix
locationService
  .listByPrefix("NFIRAOS.ncc")
  .map(akkaLocations => {
    log.info("Registered akka locations for nfiraos.ncc")
    akkaLocations.foreach(c => log.info(s"--- ${locationInfoToString(c)}"))
  })

Java

copysourceList<Location> locations = locationService.listByPrefix("NFIRAOS.ncc").get();
log.info("Registered akka locations for nfiraos.ncc");
for (Location loc : locations) {
    log.info("--- " + connectionInfo(loc.connection()));
}

The log output should contain:

[INFO] Registered akka locations for nfiraos.ncc
[INFO] --- nfiraos.ncc.hcd1-hcd-akka, component type=HCD, connection type=AkkaType
[INFO] --- nfiraos.ncc.assembly1-assembly-akka, component type=Assembly, connection type=AkkaType

Tracking and Subscribing

The lifecycle of a connection of interest can be followed using either the track API or the subscribe API.

These methods take a Connection instance as a parameter. A Connection need not already be registered with Location Service. It’s okay to track connections that will be registered in future.

The track API returns two values:
* A source that will emit a stream of TrackingEvents for the connection.
* A KillSwitch to turn off the stream when no longer needed.

The Akka stream API provides many building blocks to process this stream, such as Flow and Sink. In the example below, Sink.actorRef is used to forward any location messages received to the current actor (self).

A consumer can shut down the stream using the KillSwitch.

The subscribe API allows the caller to track a connection and receive the TrackingEvent notifications via a callback.

The API expects following parameters:
* An existing connection or a connection to be registered in the future.
* A callback that implements Consumer and receives the TrackEvent as a parameter.

This API returns a KillSwitch that can be used to turn off the event notifications and release the supplied callback, if required.

Scala

copysourcedef sinkBehavior: Behaviors.Receive[ExampleMessages] =
  Behaviors.receive[ExampleMessages] { (ctx, msg) =>
    {
      val log: Logger = new LoggerFactory(Prefix("csw.my-component-name")).getLogger(ctx)

      msg match {
        case TrackingEventAdapter(LocationUpdated(loc)) => log.info(s"Location updated ${locationInfoToString(loc)}")
        case TrackingEventAdapter(LocationRemoved(conn)) =>
          log.warn(s"Location removed $conn", Map("connection" -> conn.toString))
        case AllDone(exampleConnection) => log.info(s"Tracking of $exampleConnection complete.")
        case CustomException(throwable) => log.error(throwable.getMessage, ex = throwable)
      }
      Behaviors.same
    }
  }
  // the following two methods are examples of two ways to track a connection.
  // both are implemented but only one is really needed.

  // Method1: track connection to LocationServiceExampleComponent
  // Calls track method for example connection and forwards location messages to this actor
  //
  log.info(s"Starting to track $exampleConnection")
  val sinfActorRef = typedSystem.spawn(LocationServiceExampleClient.sinkBehavior, "")
  locationService
    .track(exampleConnection)
    .map(TrackingEventAdapter)
    .to(ActorSink.actorRef[ExampleMessages](sinfActorRef, AllDone(exampleConnection), CustomException))
    .run()
  // track returns a Killswitch, that can be used to turn off notifications arbitarily
  // in this case track a connection for 5 seconds, after that schedule switching off the stream
  val killswitch = locationService
    .track(httpConnection)
    .toMat(Sink.foreach(println))(Keep.left)
    .run()
  context.system.scheduler.scheduleOnce(5.seconds) {
    killswitch.cancel()
  }

  // Method2: subscribe to LocationServiceExampleComponent events
  log.info(s"Starting a subscription to $exampleConnection")
  locationService.subscribe(
    exampleConnection,
    trackingEvent => {
      // the following println is to distinguish subscription events from tracking events
      log.info("subscription event")
      self ! trackingEvent
    }
  )

Java

copysource// the following two methods are examples of two ways to track a connection.
// both are implemented but only one is really needed.

// track connection to LocationServiceExampleComponent
// Calls track method for example connection and forwards location messages to this actor
log.info("Starting to track " + exampleConnection);
locationService.track(exampleConnection).toMat(Sink.actorRef(getSelf(), AllDone.class), Keep.both()).run(typedSystem);

//track returns a Killswitch, that can be used to turn off notifications arbitarily
//in this case track a connection for 5 seconds, after that schedule switching off the stream
Pair pair = locationService.track(exampleConnection).toMat(Sink.ignore(), Keep.both()).run(typedSystem);
context().system().scheduler().scheduleOnce(
        Duration.ofSeconds(5),
        () -> ((KillSwitch) pair.first()).shutdown(),
        context().system().dispatcher());

// subscribe to LocationServiceExampleComponent events
log.info("Starting a subscription to " + exampleConnection);
locationService.subscribe(exampleConnection, trackingEvent -> {
    log.info("subscription event");
    getSelf().tell(trackingEvent, ActorRef.noSender());
});

The log output should contain the following:

[INFO] Starting to track AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly)) LocationUpdated(HttpLocation(HttpConnection(ComponentId(configuration,Service)),http://131.215.210.170:8080/path123))
[INFO] Starting a subscription to AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly))
[INFO] Subscribing to connection LocationServiceExampleComponent-assembly-akka
[INFO] Location updated LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType
[INFO] subscription event
[INFO] Location updated LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType

If you now stop the LocationServiceExampleComponentApp, it would log:

[INFO] subscription event
[INFO] Location removed AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly))
[INFO] Location removed AkkaConnection(ComponentId(LocationServiceExampleComponent,Assembly))

If you start the LocationServiceExampleComponentApp again, the log output should be:

[INFO] subscription event
[INFO] Location updated LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType
[INFO] Location updated LocationServiceExampleComponent-assembly-akka, component type=Assembly, connection type=AkkaType

Note: the line after the words “subscription event” in our example is generated by the subscription, and the other line is from tracking. These two events could come in any order.

Unregistering

One of the ways to unregister a service is by calling unregister on registration result received from register API.

Scala

copysourceval unregisterF = async {
  await(httpRegResultF.map(_.unregister()))
  await(httpRegResultOnPublicNetworkF.map(_.unregister()))
  await(hcdRegResultF.map(_.unregister()))
  await(assemblyRegResultF.map(_.unregister()))
}.flatten
val eventualDone = async {
  await(unregisterF)

Java

copysourcehttpRegResult.unregister();
httpRegResultonPublicNetwork.unregister();
hcdRegResult.unregister();
assemblyRegResult.unregister();

Protected Routes

The following Location Server routes are Protected. To use these routes, the user must be authenticated and authorized with location-admin role.

• register
• unregister
• unregisterAll

Technical Description

See Location Service Technical Description.

Source code for examples

• Scala Example
• JavaBlocking Example