The YARN Service Registry

Introduction and concepts

This document describes a Hadoop service registry built to address two problems:

1. How can clients talk to YARN-deployed services and the components which form such services?
2. Allow Hadoop core services to be registered and discovered thereby reducing configuration parameters and to allow core services to be more easily moved.

Service registration and discovery is a long-standing problem in distributed computing, dating back to Xerox’s Grapevine Service. This proposal is for a registry for locating distributed applications deployed by YARN, and determining the binding information needed to communicate with these applications.

The binding problem

Hadoop YARN allows applications to run on the Hadoop cluster. Some of these are batch jobs or queries that can managed via YARN’s existing API using its application ID. In addition YARN can deploy long-lived services instances such a pool of Apache Tomcat web servers or an Apache HBase cluster. YARN will deploy them across the cluster depending on the individual each component requirements and server availability. These service instances need to be discovered by clients; traditionally their IP added is registered in DNS or in some configuration file —but that is not feasible in YARN-deployed applications when neither the hostname nor network ports can be known in advance.

As a result there is no easy way for clients to interact with dynamically deployed applications.

YARN supports a rudimentary registry which allows YARN Application Masters to register a web URL and an IPC address. but is not sufficient for our purposes since it It does not allow any other endpoints to be registered —such as REST URLs, or zookeeper path or the endpoints of the tasks that the Application Master executes. Further, information that can be registered is mapped to the YARN application instance —a unique instance ID that changes whenever a YARN application is started. This makes it impossible to resolve binding information via a static reference to a named service, or to even probe for the existence of a service instance which is not currently live.

Use Cases

Service Name Examples

Core Hadoop services.

These may be deployed statically, dynamically via an account with the permissions to write to the /services path, or even registrations of remote services accessible from within the Hadoop cluster

    /services/hdfs
    /services/yarn
    /services/oozie

YARN-deployed services belonging to individual users.

    /users/joe/org-apache-hbase/demo1
    /users/joe/org-apache-hbase/demo1/components/regionserver1

Registration Use Cases

1. A Hadoop core service that is not running under YARN example: HDFS) can be registered in for discovery. This could be done by the service or by management tools..

2. A long-lived application deployed by YARN registers itself for discovery by clients. The registration data is intended to outlive the application master, and perhaps the lifespan of a single deployment of the service instance.

3. Component instances of a service register themselves, publishing internal binding information, such as JMX ports.

4. A YARN-deployed application can bind to dependent service instances both static and dynamic. Example: a Tomcat web pool binding to the dynamic HBase service instance “/users/joe/services/hbase/demo1”.

5. Component Instances use the registry to bind to an internal endpoint of their application master, to which they heartbeat regularly.

Unsupported Registration use cases:

1. A short-lived YARN application is registered automatically in the registry, including all its containers. and unregistered when the job terminates. Short-lived applications with many containers will place excessive load on a registry. All YARN applications will be given the option of registering, but it will not be automatic —and application authors must be advised against registering short-lived containers.

Lookup Use Cases

1. A client application looks up a dynamically deployed service instance whose user, service class and instance name is known, e.g. /users/joe/services/hbase/demo1, and retrieves the information needed to connect to the service

2. A client application looks up a statically deployed Hadoop service Example: /services/hdfs.

3. An Application Master enumerates all registered component instances, discovers their listed JMX ports, and, initializes own web UI, offers links to these endpoints.

4. A user connects to a private HBase service instance at /users/joe/services/hbase/demo1.

5. A user connects to the cluster’s HBase service at /services/hbase.

6. A user looks up the binding information to a remote Hadoop cluster’s filesystem at /net/cluster4/services/hdfs. The registration information includes the webhdfs:// URL for the remote filesystem.

7. A user lists their HBase service instances:

   ls /users/joe/services/hbase
   

8. User finds all Hbase services in the cluster:

   find -endpointField.api=org.apache.hbase
   

9. Possibly in future: looking up a service via DNS.

This registry proposal is intended to support these use cases by providing a means for applications to register their service endpoints, and for clients to locate them.

Key Requirements of a Service Registry

Allow dynamic registration of service instances

• YARN deployed services instances must be able register their bindings and be discovered by clients.

• Core Hadoop service instances must be able to register their service endpoints.

• The binding must be upgradable if the service moves or in case if HA fails over.

• A service instance must be able to publish a variety of endpoints for a service: Web UI, RPC, REST, Zookeeper, others. Furthermore one must also be able register certificates and other public security information may be published as part of a binding.

Registry service properties:

• The registry must be highly available.

• Scale: many services and many clients in a large cluster. This will limit how much data a service can publish.

• Ubiquity: we need this in every YARN cluster, whether physical, virtual or in-cloud.

• Must support hierarchical namespace and names. The name convention must match that of DNS so that we have the option of accessing the namespace via DNS protocol at a later phase of the project.

• Registry API Language/Protocols

• Cross-language: independent of any language; client language != service

• REST API for reading registry data

Access Control:

• Read access for all

• Write is restricted so that squatting and impersonation can be avoided.

Remote accessibility: supports remote access even on clusters which are only reachable via Apache Knox, or hosted in cloud environments.

Non-Requirements

• The registry is not intended for liveness detection, leader-election or perform other “shared consensual state” actions for an application itself, other than potentially sharing binding information between component instances.

• The registry is not intended to be a store for arbitrary application state, or for publishing configuration data other than binding information to endpoints offered by a service and its components. Such use would overload the registry and rapidly reach limits of what Zookeeper permits.

Architecture

We propose a base registry service that binds string-names to records describing service and component instances. We plan to use ZK as the base name service since it supports many of the properties, We pick a part of the ZK namespace to be the root of the service registry ( default: yarnRegistry).

On top this base implementation we build our registry service API and the naming conventions that YARN will use for its services. The registry will be accessed by the registry API, not directly via ZK - ZK is just an implementation choice (although unlikely to change in the future).

1. Services are registered by binding a path to a value called a Service Record. Paths are hierarchical and use / as the root as well as the separator.

2. Service records are registered as persistent znodes. This ensures that the record remains present during planned and unplanned outages of the service, on the assumption that client code is resilient to transient outages.

3. Each service instance’s service record lists the endpoints for its various protocols exported by that service instance.

4. For each protocol endpoint it must contain

   1. The protocol name including: Web, REST, IPC, zookeeper. (type:string)

   2. Its address: the specific details used to locate this endpoint

   3. Its addressType. This is the format of the binding string. (URL, ZK path, hostname:port pair). For the predefined protocols, we will define what format the binding string MUST be. Example: protocol==REST means binding type is URL, protocol==IPC binding uses the addresstype host/port.

   4. The api. This is the API offered by the endpoint, and is application specific. examples: org.apache.hadoop.namenode, org.apache.hadoop.webhdfs

5. Endpoints may be external —for use by programs other than the service itself, and internal —for connecting components within the service instance. They will be listed in different sections of the Service Record to distinguish them.

6. Core services will be registered using the following convention: /services/{servicename} e.g. /services/hdfs.

7. YARN services SHOULD be registered using the following convention:

   /users/{username}/{serviceclass}/{instancename}
   

8. Component instances SHOULD be registered under

   /users/{username}/{serviceclass}/{instancename}/components/{componentname}
   

9. Each of the user’s services which follows this convention must have unique service class names,

10. Each component instance must have a name that is unique for that service instance. For a YARN-deployed application, this can be trivially derived from the container ID.

The requirements for unique names ensures that the path to a service instance or component instance is guaranteed to be unique, and that all instances of a specific service class can be enumerated by listing all children of the service class path.

Registry Model

Service entries MUST be persistent —it is the responsibility of YARN and other tools to determine when a service entry is to be deleted.

Path Elements

All path elements MUST match that of a lower-case entry in a hostname path as defined in RFC1123; the regular expression is:

([a-z0-9]|([a-z0-9][a-z0-9\-]*[a-z0-9]))

This policy will ensure that were the registry hierarchy ever to exported by a DNS service, all service classes and names would be valid.

A complication arises with user names, as platforms may allow user names with spaces, high unicode and other characters in them. Such paths must be converted to valid DNS hostname entries using the punycode convention used for internationalized DNS.

Service Record

A Service Record has some basic information and possibly empty lists of internal and external endpoints.

/users
  /devteam
   /org-apache-tomcat
     /test1
       /components
         /container-1408631738011-0001-01-000002
         /container-1408631738011-0001-01-000001

{
  "description" : "tomcat-based web application",
  "external" : [ {
    "api" : "http://internal.example.org/restapis/scheduler/20141026v1",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [
     { "uri" : "http://loadbalancer/" },
     { "uri" : "http://loadbalancer2/" }
      ]
  } ],
  "internal" : [ ]
}

/users/devteam/org-apache-tomcat/test1/components/container-1408631738011-0001-01-000001

{
  "yarn:id" : "container_1408631738011_0001_01_000001",
  "yarn:persistence" : "container",
  "description" : "",
  "external" : [ {
    "api" : "http://internal.example.org/restapis/scheduler/20141026v1",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [{ "uri" : "rack4server3:43572" }  ]
  } ],
  "internal" : [ {
    "api" : "classpath:javax.management.jmx",
    "addressType" : "host/port",
    "protocol" : "rmi",
    "addresses" : [ {
      "host" : "rack4server3",
      "port" : "48551"
    } ]
  } ]
}

/users/devteam/org-apache-tomcat/test1/components/container-1408631738011-0001-01-000002

{
  "registrationTime" : 1408638082445,
  "yarn:id" : "container_1408631738011_0001_01_000002",
  "yarn:persistence" : "container",
  "description" : null,
  "external" : [ {
    "api" : "http://internal.example.org/restapis/scheduler/20141026v1",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ [ "http://rack1server28:35881" ] ]
  } ],
  "internal" : [ {
    "api" : "classpath:javax.management.jmx",
    "addressType" : "host/port",
    "protocol" : "rmi",
    "addresses" : [ {
      "host" : "rack1server28",
      "port" : "48551"
    } ]
  } ]
}

public interface RegistryOperations extends Service {

  /**
   * Create a path.
   *
   * It is not an error if the path exists already, be it empty or not.
   *
   * The createParents flag also requests creating the parents.
   * As entries in the registry can hold data while still having
   * child entries, it is not an error if any of the parent path
   * elements have service records.
   *
   * @param path path to create
   * @param createParents also create the parents.
   * @throws PathNotFoundException parent path is not in the registry.
   * @throws InvalidPathnameException path name is invalid.
   * @throws IOException Any other IO Exception.
   * @return true if the path was created, false if it existed.
   */
  boolean mknode(String path, boolean createParents)
      throws PathNotFoundException,
      InvalidPathnameException,
      IOException;

  /**
   * Set a service record to an entry
   * @param path path to service record
   * @param record service record service record to create/update
   * @param createFlags creation flags
   * @throws PathNotFoundException the parent path does not exist
   * @throws FileAlreadyExistsException path exists but create flags
   * do not include "overwrite"
   * @throws InvalidPathnameException path name is invalid.
   * @throws IOException Any other IO Exception.
   */
  void bind(String path, ServiceRecord record, int createFlags)
      throws PathNotFoundException,
      FileAlreadyExistsException,
      InvalidPathnameException,
      IOException;

  /**
   * Resolve the record at a path
   * @param path path to service record
   * @return the record
   * @throws PathNotFoundException path is not in the registry.
   * @throws InvalidPathnameException the path is invalid.
   * @throws IOException Any other IO Exception
   */

  ServiceRecord resolve(String path) throws PathNotFoundException,
      InvalidPathnameException,
      IOException;

  /**
   * Get the status of a path
   * @param path path to query
   * @return the status of the path
   * @throws PathNotFoundException path is not in the registry.
   * @throws InvalidPathnameException the path is invalid.
   * @throws IOException Any other IO Exception
   */
  RegistryPathStatus stat(String path)
      throws PathNotFoundException,
      InvalidPathnameException,
      IOException;

  /**
   * Probe for a path existing.
   * This is equivalent to {@link #stat(String)} with
   * any failure downgraded to a
   * @param path path to query
   * @return true if the path was found
   * @throws IOException
   */
  boolean exists(String path) throws IOException;

 /**
   * List all entries under a registry path
   * @param path path to query
   * @return a possibly empty list of the full path names of
   * child entries
   * @throws PathNotFoundException
   * @throws InvalidPathnameException
   * @throws IOException
   */
   List<String> list(String path) throws
      PathNotFoundException,
      InvalidPathnameException,
      IOException;

  /**
   * Delete a path.
   *
   * If the operation returns without an error then the entry has been
   * deleted.
   * @param path path delete recursively
   * @param recursive recursive flag
   * @throws PathNotFoundException path is not in the registry.
   * @throws InvalidPathnameException the path is invalid.
   * @throws PathIsNotEmptyDirectoryException path has child entries, but
   * recursive is false.
   * @throws IOException Any other IO Exception
   *
   */
  void delete(String path, boolean recursive)
      throws PathNotFoundException,
      PathIsNotEmptyDirectoryException,
      InvalidPathnameException,
      IOException;

  /**
   * Add a new write access entry to be added to node permissions in all
   * future write operations of a session connected to a secure registry.
   *
   * This does not grant the session any more rights: if it lacked any write
   * access, it will still be unable to manipulate the registry.
   *
   * In an insecure cluster, this operation has no effect.
   * @param id ID to use
   * @param pass password
   * @return true if the accessor was added: that is, the registry connection
   * uses permissions to manage access
   * @throws IOException on any failure to build the digest
   */
  boolean addWriteAccessor(String id, String pass) throws IOException;

  /**
   * Clear all write accessors.
   *
   * At this point all standard permissions/ACLs are retained,
   * including any set on behalf of the user
   * Only  accessors added via {@link #addWriteAccessor(String, String)}
   * are removed.
   */
  public void clearWriteAccessors();
}

public class RegistryPathStatus {

  /**
   * Short path in the registry to this entry
   */
  public String path;

  /**
   * Timestamp
   */
  public long time;

  /**
   * Entry size in bytes, as returned by the storage infrastructure.
   * In zookeeper, even "empty" nodes have a non-zero size.
   */
  public long size;

  /**
   * Number of child nodes
   */
  public int children;
}

Security

The registry will allow a service instance can only be registered under the path where it has permissions. YARN will create directories with appropriate permissions for users where YARN deployed services can be registered by a user. of the user account of the service instance. The admin will also create directories (such as /services) with appropriate permissions (where core Hadoop services can register themselves.

There will no attempt to restrict read access to registry information. The services will protect inappropriate access by clients by requiring authentication and authorization. There is a scope field in a service record , but this is just a marker to say “internal API only”, rather than a direct security restriction. (this is why “internal” and “external” are proposed, not “public” and “private”).

Rationale: the endpoints being registered would be discoverable through port scanning anyway. Having everything world-readable allows the REST API to have a simpler access model —and is consistent with DNS.

On a secure cluster, ZK token renewal may become an issue for long-lived services —if their token expires their session may expire. Renewal of such tokens is not part of the API implementation —we may need to add a means to update the tokens of an instance of the registry operations class.

Out of cluster and cross-cluster access

1. A client should be able to access the registry of another cluster in order to access services of that cluster. Detail of this need to further fleshed out.

2. Firewall services such as Apache Knox can examine the internal set of published services, and publish a subset of their endpoints. They MAY implement a future REST API.

Limits

Entry Size

Zookeeper has a default limit of 1MB/node. If all endpoints of a service or component are stored in JSON attached to that node, then there is a total limit of 1MB of all endpoint registration data.

To prevent this becoming a problem, the client API should implement strict limits on the maximum length of fields, with low limits on the addressType, protocol, and api fields, something longer on the description and addresses elements —along with a limit on the number of elements in the addresses field.

Name size

To support DNS in future, there must be a limit of 63 bytes on all path elements. For non-ASCII User names, this restriction implies that a shorter path may be a limit.

Rate of Update

A rapid rate of entry change is considered antisocial in a ZK cluster. Implementations may throttle update operations.

Rate of Polling

Clients which poll the registry may be throttled.

Complete service record example

Below is a (non-normative) example of a service record retrieved from a YARN application.

{
  "type" : "JSONServiceRecord",
  "description" : "Slider Application Master",
  "yarn:persistence" : "application",
  "yarn:id" : "application_1414052463672_0028",
  "external" : [ {
    "api" : "classpath:org.apache.slider.appmaster",
    "addressType" : "host/port",
    "protocol" : "hadoop/IPC",
    "addresses" : [ {
      "port" : "48551",
      "host" : "nn.example.com"
    } ]
  }, {
    "api" : "http://",
    "addressType" : "uri",
    "protocol" : "web",
    "addresses" : [ {
      "uri" : "http://nn.example.com:40743"
    } ]
  }, {
    "api" : "classpath:org.apache.slider.management",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "http://nn.example.com:40743/ws/v1/slider/mgmt"
    } ]
  }, {
    "api" : "classpath:org.apache.slider.publisher",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "http://nn.example.com:40743/ws/v1/slider/publisher"
    } ]
  }, {
    "api" : "classpath:org.apache.slider.registry",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "http://nn.example.com:40743/ws/v1/slider/registry"
    } ]
  }, {
    "api" : "classpath:org.apache.slider.publisher.configurations",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "http://nn.example.com:40743/ws/v1/slider/publisher/slider"
    } ]
  }, {
    "api" : "classpath:org.apache.slider.publisher.exports",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "http://nn.example.com:40743/ws/v1/slider/publisher/exports"
    } ]
  } ],
  "internal" : [ {
    "api" : "classpath:org.apache.slider.agents.secure",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "https://nn.example.com:52705/ws/v1/slider/agents"
    } ]
  }, {
    "api" : "classpath:org.apache.slider.agents.oneway",
    "addressType" : "uri",
    "protocol" : "REST",
    "addresses" : [ {
      "uri" : "https://nn.example.com:33425/ws/v1/slider/agents"
    } ]
  } ]
}

It publishes a number of endpoints, both internal and external.

External:

1. The IPC hostname and port for client-AM communications
2. URL to the AM’s web UI
3. A series of REST URLs under the web UI for specific application services. The details are irrelevant —note that they use an application-specific API value to ensure uniqueness.

Internal:

1. Two URLs to REST APIs offered by the AM for containers deployed by the application itself.

Python agents running in the containers retrieve the internal endpoint URLs to communicate with their AM. The record is resolved on container startup and cached until communications problems occur. At that point the registry is queried for the current record, then an attempt is made to reconnect to the AM.

Here “connectivity” problems means both “low level socket/IO errors” and “failures in HTTPS authentication”. The agents use two-way HTTPS authentication —if the AM fails and another application starts listening on the same ports it will trigger an authentication failure and hence service record reread.