Skip to content

[LIVY-41] Let users access sessions by session name #48

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
meisam wants to merge 25 commits into from
Closed

[LIVY-41] Let users access sessions by session name #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
8487bbb
[LIVY-41] Let users access sessions by session name
meisam Sep 12, 2017
3f28b61
[LIVY-41] renaming id tp idOrName
meisam Oct 5, 2017
dc644c7
[LIVY-41] refactoring guareded case statements
meisam Oct 5, 2017
967e896
First shot at making brnach names optional
meisam Oct 10, 2017
0a0e1cf
[LIVY-41] Making session names optional
meisam Oct 16, 2017
c16685a
Change name from null to None in test cases
meisam Jan 14, 2019
c34eca8
Validate session names
meisam Jan 14, 2019
5cbef17
Merge remote-tracking branch 'apache/master' into LIVY-41
meisam Jan 15, 2019
c995ab2
Exclude numbers from valid session names.
meisam Jan 15, 2019
9f3915d
Cosmetics and code formatting
meisam Jan 15, 2019
ecde1fd
Set Thriftserver session names to None
meisam Jan 15, 2019
57ec81a
change method calls from infix form to . form in InteractiveSession
meisam Jan 15, 2019
2065619
Test sessions without a name can be recovered.
meisam Jan 22, 2019
962e82e
Cosmetics and code formatting
meisam Jan 26, 2019
3527ef7
Simplify check for session names.
meisam Jan 26, 2019
9c64c81
Start heartbeat and save session recovery data after start() is called.
meisam Jan 26, 2019
516a880
Inline check for session names.
meisam Jan 30, 2019
334b165
Refactor common test case code snippets into helper functions
meisam Jan 30, 2019
ef56dea
Decode string bytes as "UTF-8"
meisam Jan 30, 2019
41ff9aa
Remove the irrelevant checks for the duplicate session names test
meisam Jan 30, 2019
d673baf
Guard if caluse with braces
meisam Jan 30, 2019
a43a968
[LIVY-41] Fix typo : "UTF_8" -> "UTF-8"
meisam Jan 30, 2019
35c34d7
[LIVY-41] Fix tests for session names
meisam Jan 30, 2019
518bfdf
[LIVY-41] Remove ID parameter from testShowSessionProperties
meisam Jan 30, 2019
f61c3d5
Fix the test cases after refactoring
meisam Jan 31, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions client-common/src/main/java/org/apache/livy/client/common/HttpMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ private CreateClientRequest() {
public static class SessionInfo implements ClientMessage {

public final int id;
public final String name;
public final String appId;
public final String owner;
public final String proxyUser;
Expand All @@ -61,9 +62,10 @@ public static class SessionInfo implements ClientMessage {
public final Map<String, String> appInfo;
public final List<String> log;

public SessionInfo(int id, String appId, String owner, String proxyUser, String state,
String kind, Map<String, String> appInfo, List<String> log) {
public SessionInfo(int id, String name, String appId, String owner, String proxyUser,
String state, String kind, Map<String, String> appInfo, List<String> log) {
this.id = id;
this.name = name;
this.appId = appId;
this.owner = owner;
this.proxyUser = proxyUser;
Expand All @@ -74,7 +76,7 @@ public SessionInfo(int id, String appId, String owner, String proxyUser, String
}

private SessionInfo() {
this(-1, null, null, null, null, null, null, null);
this(-1, null, null, null, null, null, null, null, null);
}

}
Expand Down
1 change: 1 addition & 0 deletions client-http/src/test/scala/org/apache/livy/client/http/HttpClientSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,7 @@ private class HttpClientTestBootstrap extends LifeCycle {
val session = mock(classOf[InteractiveSession])
val id = sessionManager.nextId()
when(session.id).thenReturn(id)
when(session.name).thenReturn(None)
when(session.appId).thenReturn(None)
when(session.appInfo).thenReturn(AppInfo())
when(session.state).thenReturn(SessionState.Idle)
Expand Down
4 changes: 4 additions & 0 deletions integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,12 +249,14 @@ class LivyRestClient(val httpClient: AsyncHttpClient, val livyEndpoint: String)
}

def startBatch(
name: Option[String],
file: String,
className: Option[String],
args: List[String],
sparkConf: Map[String, String]): BatchSession = {
val r = new CreateBatchRequest()
r.file = file
r.name = name
r.className = className
r.args = args
r.conf = Map("spark.yarn.maxAppAttempts" -> "1") ++ sparkConf
Expand All @@ -264,12 +266,14 @@ class LivyRestClient(val httpClient: AsyncHttpClient, val livyEndpoint: String)
}

def startSession(
name: Option[String],
kind: Kind,
sparkConf: Map[String, String],
heartbeatTimeoutInSecond: Int): InteractiveSession = {
val r = new CreateInteractiveRequest()
r.kind = kind
r.conf = sparkConf
r.name = name
r.heartbeatTimeoutInSecond = heartbeatTimeoutInSecond

val id = start(INTERACTIVE_TYPE, mapper.writeValueAsString(r))
Expand Down
4 changes: 2 additions & 2 deletions integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,14 +159,14 @@ class BatchIT extends BaseIntegrationTestSuite with BeforeAndAfterAll {
private def withScript[R]
(scriptPath: String, args: List[String], sparkConf: Map[String, String] = Map.empty)
(f: (LivyRestClient#BatchSession) => R): R = {
val s = livyClient.startBatch(scriptPath, None, args, sparkConf)
val s = livyClient.startBatch(None, scriptPath, None, args, sparkConf)
withSession(s)(f)
}

private def withTestLib[R]
(testClass: Class[_], args: List[String], sparkConf: Map[String, String] = Map.empty)
(f: (LivyRestClient#BatchSession) => R): R = {
val s = livyClient.startBatch(testLibPath, Some(testClass.getName()), args, sparkConf)
val s = livyClient.startBatch(None, testLibPath, Some(testClass.getName()), args, sparkConf)
withSession(s)(f)
}
}
2 changes: 1 addition & 1 deletion integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ class InteractiveIT extends BaseIntegrationTestSuite {
waitForIdle: Boolean = true,
heartbeatTimeoutInSecond: Int = 0)
(f: (LivyRestClient#InteractiveSession) => R): R = {
withSession(livyClient.startSession(kind, sparkConf, heartbeatTimeoutInSecond)) { s =>
withSession(livyClient.startSession(None, kind, sparkConf, heartbeatTimeoutInSecond)) { s =>
if (waitForIdle) {
s.verifySessionIdle()
}
Expand Down
13 changes: 10 additions & 3 deletions server/src/main/scala/org/apache/livy/server/SessionServlet.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,16 +184,23 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
private def doWithSession(fn: (S => Any),
allowAll: Boolean,
checkFn: Option[(String, String) => Boolean]): Any = {
val sessionId = params("id").toInt
sessionManager.get(sessionId) match {
val idOrNameParam: String = params("id")
val session = if (idOrNameParam.forall(_.isDigit)) {

@vanzin vanzin Dec 19, 2018

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're assuming that a string with all digits is an ID and not a name. I don't see that enforced when creating a session?

@meisam meisam Jan 15, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I changed this to force session names to have at least one non-digit character.

val sessionId = idOrNameParam.toInt
sessionManager.get(sessionId)
} else {
val sessionName = idOrNameParam
sessionManager.get(sessionName)
}
session match {
case Some(session) =>
if (allowAll || checkFn.map(_(session.owner, remoteUser(request))).getOrElse(false)) {
fn(session)
} else {
Forbidden()
}
case None =>
NotFound(ResponseMessage(s"Session '$sessionId' not found."))
NotFound(ResponseMessage(s"Session '$idOrNameParam' not found."))
}
}

Expand Down
20 changes: 15 additions & 5 deletions server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import org.apache.livy.utils.{AppInfo, SparkApp, SparkAppListener, SparkProcessB
@JsonIgnoreProperties(ignoreUnknown = true)
case class BatchRecoveryMetadata(
id: Int,
name: Option[String],
appId: Option[String],
appTag: String,
owner: String,
Expand All @@ -53,6 +54,7 @@ object BatchSession extends Logging {

def create(
id: Int,
name: Option[String],
request: CreateBatchRequest,
livyConf: LivyConf,
accessManager: AccessManager,
Expand Down Expand Up @@ -110,6 +112,7 @@ object BatchSession extends Logging {

new BatchSession(
id,
name,
appTag,
SessionState.Starting,
livyConf,
Expand All @@ -126,6 +129,7 @@ object BatchSession extends Logging {
mockApp: Option[SparkApp] = None): BatchSession = {
new BatchSession(
m.id,
m.name,
m.appTag,
SessionState.Recovering,
livyConf,
Expand All @@ -140,27 +144,33 @@ object BatchSession extends Logging {

class BatchSession(
id: Int,
name: Option[String],
appTag: String,
initialState: SessionState,
livyConf: LivyConf,
owner: String,
override val proxyUser: Option[String],
sessionStore: SessionStore,
sparkApp: BatchSession => SparkApp)
extends Session(id, owner, livyConf) with SparkAppListener {
extends Session(id, name, owner, livyConf) with SparkAppListener {
import BatchSession._

protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global

private[this] var _state: SessionState = initialState
private val app = sparkApp(this)

private var app: Option[SparkApp] = None

override def state: SessionState = _state

override def logLines(): IndexedSeq[String] = app.log()
override def logLines(): IndexedSeq[String] = app.map(_.log()).getOrElse(IndexedSeq.empty[String])

override def start(): Unit = {
app = Option(sparkApp(this))
}

override def stopSession(): Unit = {
app.kill()
app.foreach(_.kill())
}

override def appIdKnown(appId: String): Unit = {
Expand All @@ -187,5 +197,5 @@ class BatchSession(
override def infoChanged(appInfo: AppInfo): Unit = { this.appInfo = appInfo }

override def recoveryMetadata: RecoveryMetadata =
BatchRecoveryMetadata(id, appId, appTag, owner, proxyUser)
BatchRecoveryMetadata(id, name, appId, appTag, owner, proxyUser)

@jerryshao jerryshao Sep 28, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@meisam what if old recovery metadata which doesn't have name field, I think we should consider the compatibility of old metadata file with new Livy server, would you please check it?

@meisam meisam Sep 29, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you suggest? Should we assign such sessions auto-generated names on recovery? Or should we leave them without a name?

@jerryshao jerryshao Sep 30, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm saying with your changes, if new Livy server tries to recover from old data, what will be happened? Will it be failed?

Since the old recovery data doesn't have a session name field, so we could offer an auto-generated name, but we should not fail such scenario.

@meisam meisam Jan 15, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll add test cases to confirm that Livy can recover from old data.

}
9 changes: 7 additions & 2 deletions server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import org.apache.livy.utils.AppInfo

case class BatchSessionView(
id: Long,
name: Option[String],
state: String,
appId: Option[String],
appInfo: AppInfo,
Expand All @@ -42,8 +43,11 @@ class BatchSessionServlet(

override protected def createSession(req: HttpServletRequest): BatchSession = {
val createRequest = bodyAs[CreateBatchRequest](req)
val sessionId = sessionManager.nextId()
val sessionName = createRequest.name
BatchSession.create(
sessionManager.nextId(),
sessionId,
sessionName,
createRequest,
livyConf,
accessManager,
Expand All @@ -66,7 +70,8 @@ class BatchSessionServlet(
} else {
Nil
}
BatchSessionView(session.id, session.state.toString, session.appId, session.appInfo, logs)
BatchSessionView(session.id, session.name, session.state.toString, session.appId,
session.appInfo, logs)
}

}
Loading