Skip to content

Commit

Permalink
Initial client, server, and all the messages
Browse files Browse the repository at this point in the history
This commit introduces type-safe schemas for all messages exchanged
in the REST protocol. Each message is expected to contain an ACTION
field that has only one possible value for each message type.

Before the message is sent, we validate that all required fields
are in fact present, and that the value of the action field is the
correct type.

The next step is to actually integrate this in standalone mode.
  • Loading branch information
Andrew Or committed Jan 17, 2015
1 parent 14e3f11 commit 53e7c0e
Show file tree
Hide file tree
Showing 11 changed files with 910 additions and 2 deletions.
5 changes: 3 additions & 2 deletions core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ object SparkSubmit {
private val STANDALONE = 2
private val MESOS = 4
private val LOCAL = 8
private val ALL_CLUSTER_MGRS = YARN | STANDALONE | MESOS | LOCAL
private val REST = 16
private val ALL_CLUSTER_MGRS = YARN | STANDALONE | MESOS | LOCAL | REST

// Deploy modes
private val CLIENT = 1
Expand Down Expand Up @@ -97,7 +98,7 @@ object SparkSubmit {
case m if m.startsWith("spark") => STANDALONE
case m if m.startsWith("mesos") => MESOS
case m if m.startsWith("local") => LOCAL
case _ => printErrorAndExit("Master must start with yarn, spark, mesos, or local"); -1
case _ => printErrorAndExit("Master must start with yarn, spark, mesos, local, or rest"); -1
}

// Set the deploy mode; default is client mode
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.spark.deploy.rest

/**
* A field used in a DriverStatusRequestMessage.
*/
private[spark] abstract class DriverStatusRequestField extends StandaloneRestProtocolField
private[spark] object DriverStatusRequestField extends StandaloneRestProtocolFieldCompanion {
case object ACTION extends DriverStatusRequestField
case object SPARK_VERSION extends DriverStatusRequestField
case object MESSAGE extends DriverStatusRequestField
case object MASTER extends DriverStatusRequestField
case object DRIVER_ID extends DriverStatusRequestField
override val requiredFields = Seq(ACTION, SPARK_VERSION, MASTER, DRIVER_ID)
override val optionalFields = Seq(MESSAGE)
}

/**
* A request sent to the standalone Master to query the status of a driver.
*/
private[spark] class DriverStatusRequestMessage extends StandaloneRestProtocolMessage(
StandaloneRestProtocolAction.DRIVER_STATUS_REQUEST,
DriverStatusRequestField.ACTION,
DriverStatusRequestField.requiredFields)

private[spark] object DriverStatusRequestMessage extends StandaloneRestProtocolMessageCompanion {
protected override def newMessage(): StandaloneRestProtocolMessage =
new DriverStatusRequestMessage
protected override def fieldWithName(field: String): StandaloneRestProtocolField =
DriverStatusRequestField.withName(field)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.spark.deploy.rest

/**
* A field used in a DriverStatusResponseMessage.
*/
private[spark] abstract class DriverStatusResponseField extends StandaloneRestProtocolField
private[spark] object DriverStatusResponseField extends StandaloneRestProtocolFieldCompanion {
case object ACTION extends DriverStatusResponseField
case object SPARK_VERSION extends DriverStatusResponseField
case object MESSAGE extends DriverStatusResponseField
case object MASTER extends DriverStatusResponseField
case object DRIVER_ID extends DriverStatusResponseField
case object DRIVER_STATE extends SubmitDriverResponseField
case object WORKER_ID extends SubmitDriverResponseField
case object WORKER_HOST_PORT extends SubmitDriverResponseField
override val requiredFields = Seq(ACTION, SPARK_VERSION, MESSAGE,
MASTER, DRIVER_ID, DRIVER_STATE, WORKER_ID, WORKER_HOST_PORT)
override val optionalFields = Seq.empty
}

/**
* A message sent from the standalone Master in response to a DriverStatusResponseMessage.
*/
private[spark] class DriverStatusResponseMessage extends StandaloneRestProtocolMessage(
StandaloneRestProtocolAction.DRIVER_STATUS_RESPONSE,
DriverStatusResponseField.ACTION,
DriverStatusResponseField.requiredFields)

private[spark] object DriverStatusResponseMessage extends StandaloneRestProtocolMessageCompanion {
protected override def newMessage(): StandaloneRestProtocolMessage =
new DriverStatusResponseMessage
protected override def fieldWithName(field: String): StandaloneRestProtocolField =
DriverStatusResponseField.withName(field)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.spark.deploy.rest

/**
* A field used in a ErrorMessage.
*/
private[spark] abstract class ErrorField extends StandaloneRestProtocolField
private[spark] object ErrorField extends StandaloneRestProtocolFieldCompanion {
case object ACTION extends ErrorField
case object SPARK_VERSION extends ErrorField
case object MESSAGE extends ErrorField
override val requiredFields = Seq(ACTION, SPARK_VERSION, MESSAGE)
override val optionalFields = Seq.empty
}

/**
* An error message exchanged in the standalone REST protocol.
*/
private[spark] class ErrorMessage extends StandaloneRestProtocolMessage(
StandaloneRestProtocolAction.ERROR,
ErrorField.ACTION,
ErrorField.requiredFields)

private[spark] object ErrorMessage extends StandaloneRestProtocolMessageCompanion {
protected override def newMessage(): StandaloneRestProtocolMessage = new ErrorMessage
protected override def fieldWithName(field: String): StandaloneRestProtocolField =
ErrorField.withName(field)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.spark.deploy.rest

/**
* A field used in a KillDriverRequestMessage.
*/
private[spark] abstract class KillDriverRequestField extends StandaloneRestProtocolField
private[spark] object KillDriverRequestField extends StandaloneRestProtocolFieldCompanion {
case object ACTION extends KillDriverRequestField
case object SPARK_VERSION extends KillDriverRequestField
case object MESSAGE extends KillDriverRequestField
case object MASTER extends KillDriverRequestField
case object DRIVER_ID extends KillDriverRequestField
override val requiredFields = Seq(ACTION, SPARK_VERSION, MASTER, DRIVER_ID)
override val optionalFields = Seq(MESSAGE)
}

/**
* A request sent to the standalone Master to kill a driver.
*/
private[spark] class KillDriverRequestMessage extends StandaloneRestProtocolMessage(
StandaloneRestProtocolAction.KILL_DRIVER_REQUEST,
KillDriverRequestField.ACTION,
KillDriverRequestField.requiredFields)

private[spark] object KillDriverRequestMessage extends StandaloneRestProtocolMessageCompanion {
protected override def newMessage(): StandaloneRestProtocolMessage =
new KillDriverRequestMessage
protected override def fieldWithName(field: String): StandaloneRestProtocolField =
KillDriverRequestField.withName(field)
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.spark.deploy.rest

/**
* A field used in a KillDriverResponseMessage.
*/
private[spark] abstract class KillDriverResponseField extends StandaloneRestProtocolField
private[spark] object KillDriverResponseField extends StandaloneRestProtocolFieldCompanion {
case object ACTION extends KillDriverResponseField
case object SPARK_VERSION extends KillDriverResponseField
case object MESSAGE extends KillDriverResponseField
case object MASTER extends KillDriverResponseField
case object DRIVER_ID extends KillDriverResponseField
case object DRIVER_STATE extends SubmitDriverResponseField
override val requiredFields = Seq(ACTION, SPARK_VERSION, MESSAGE, MASTER, DRIVER_ID, DRIVER_STATE)
override val optionalFields = Seq.empty
}

/**
* A message sent from the standalone Master in response to a KillDriverResponseMessage.
*/
private[spark] class KillDriverResponseMessage extends StandaloneRestProtocolMessage(
StandaloneRestProtocolAction.KILL_DRIVER_RESPONSE,
KillDriverResponseField.ACTION,
KillDriverResponseField.requiredFields)

private[spark] object KillDriverResponseMessage extends StandaloneRestProtocolMessageCompanion {
protected override def newMessage(): StandaloneRestProtocolMessage =
new KillDriverResponseMessage
protected override def fieldWithName(field: String): StandaloneRestProtocolField =
KillDriverResponseField.withName(field)
}
Loading

0 comments on commit 53e7c0e

Please sign in to comment.