Resource Provider

gistfile1.scala

package com.typesafe.reactiveruntime.resource

import akka.actor.{ActorRef, Props, Actor}
import akka.cluster.ClusterEvent.{MemberEvent, InitialStateAsEvents, MemberRemoved, MemberUp}
import akka.cluster.{Member, Cluster, UniqueAddress}
import akka.contrib.pattern.{DistributedPubSubMediator, DistributedPubSubExtension}
import com.typesafe.reactiveruntime.resource.ResourceProvider.Event.ResourceOffer
import com.typesafe.reactiveruntime.resource.ResourceProvider.RequestResources
import java.util.UUID
import scala.collection.immutable.HashSet

object ResourceProvider {

  /**
   * Sent by a node declining a resource offer.
   * @param id the offer to decline
   * @param uniqueAddress the address of the node declining the offer
   */
  case class DeclineOffer(id: UUID, uniqueAddress: UniqueAddress)

  /**
   * Request some resources. There is no guarantee that a resource offer will be made
   * subsequently. This request is intended to provide the provider with a clue as
   * to what resources are required where.
   * @param unqiueAddress The address of the node requesting the resources
   * @param resources The resources to request
   */
  case class RequestResources(unqiueAddress: UniqueAddress, resources: List[Resource])

  object Event {

    /**
     * Offers storage. Resources offers may be made at any time.
     * @param id the id of the offer being made
     * @param unqiueAddress the node where the storage is offered
     * @param resources The resources on offer
     */
    case class ResourceOffer(id: UUID, unqiueAddress: UniqueAddress, resources: List[Resource])

  }

  sealed trait Resource

  /**
   * Storage resource.
   * @param available how much storage is available (in bytes)
   */
  case class Storage(available: Long) extends Resource

  /**
   * Name for creating a [[ResourceProvider]] actor.
   */
  val name: String =
    "resource-provider"

  /**
   * Name for distributed pub-sub topic for executor events.
   */
  val topic: String =
    "resource-provider-events"
}

/**
 * A resource manager is responsible for offering machine resources to its subscribers.
 * Resources consists of CPU, memory and storage metrics.
 */
abstract class ResourceProvider extends Actor {

  import ResourceProvider._

  protected val mediator: ActorRef =
    DistributedPubSubExtension(context.system).mediator

  protected def publish(event: ResourceOffer): Unit =
    mediator ! DistributedPubSubMediator.Publish(topic, event)
}

object SimpleResourceProvider {
  def props: Props =
    Props(new SimpleResourceProvider)
}

/**
 * A simple resource providers uses crude OS metrics to determine whether resource offers can be made.
 * While resource providers may emit resource offers at any time, this one does so only when stimulated
 * by q request for resources.
 */
class SimpleResourceProvider extends ResourceProvider {
  override def receive: Receive = {
    case request: RequestResources =>
      // FIXME: This is really dumb. We need to obtain the resources for each node and publish
      // FIXME: them appropriately.
      for {
        member <- members
      } publish(ResourceOffer(UUID.randomUUID(), member.uniqueAddress, request.resources))

    case MemberUp(member) =>
      members += member

    case MemberRemoved(member, previousStatus) =>
      members -= member
  }

  private var members: Set[Member] = HashSet.empty[Member]

  private val cluster: Cluster =
    Cluster(context.system)

  override def preStart(): Unit =
    cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberEvent])

  override def postStop(): Unit =
    cluster.unsubscribe(self)
}