mesosphere / mesos-rxjava Goto Github PK

As the Apache Mesos HTTP APIs are experimental this library should also be considered experimental. The project reserves the right to change any interface at any time until the first release is made.

This is compatible with the semver definition of the 0.y.z version numbers:

Major version zero (0.y.z) is for initial development. Anything may change at any time. The public API should not be considered stable.

So there is really no reason not to just explicitly state in README.md on this library that it follows semantic versioning and release as many incompatible 0.y.z versions as you want.

If you think about it, using SNAPSHOT versions is really the worst possible case for potential users of this library because if in my pom.xml I say "depend on 0.1.0-SNAPSHOT" there is nothing I can do to prevent two different builds my dependent project from using two different 0.1.0-SNAPSHOT (incompatible) versions.

Thanks for your consideration!

We should consider tests that run against an actual Mesos cluster vs. tests that run against a mock Mesos master.

MesosClient ignores http response content on error and doesn't report error description:

Exception in thread "main" com.mesosphere.mesos.rx.java.Mesos4xxException: Error while trying to send request. Status: 400 Message: ''
at com.mesosphere.mesos.rx.java.MesosClient.lambda$verifyResponseOk$1(MesosClient.java:260)
at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:69)
at rx.internal.operators.OperatorSubscribeOn$1$1.onNext(OperatorSubscribeOn.java:53)
at rx.observers.Subscribers$5.onNext(Subscribers.java:235)

You can find the responsible source code here:
mesos-rxjava-client-0.1.0-sources.jar!/com/mesosphere/mesos/rx/java/MesosClient.java:242

Example of the content skipped:

Failed to validate scheduler::Call: 'framework_id' differs from 'subscribe.framework_info.id'

• The variable Observable<Call> declines should be renamed to evaluatedOffers or something similar
• The offer should be declined (instead of ignored) in the else block corresponding with if (cpuList != null && !cpuList.isEmpty() && memList != null && !memList.isEmpty()).

As of 0.1.0 when a Call is process by SinkSubscriber the onComplete/onError callbacks are invoked on the netty event threads. It would be preferable to invoke the callbacks on a different thread so that we ensure the netty threads won't be blocked by what is ran in the callbacks.

Here is a log from the example framework that shows this:

> java -jar mesos-rxjava-example-framework-0.1.0-jar-with-dependencies.jar http://localhost:5050/api/v1/scheduler 2 '*'
2016-07-27 11:47:28,253 INFO  [main                     ] c.m.mesos.rx.java.MesosClient        - Probing Mesos server at http://localhost:5050/redirect
2016-07-27 11:47:28,431 INFO  [main                     ] c.m.mesos.rx.java.MesosClient        - Using new Mesos URI: http://localhost:5050/api/v1/scheduler
2016-07-27 11:47:28,791 INFO  [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.Sleepy  - Launching 2 tasks
2016-07-27 11:47:28,862 DEBUG [rxnetty-nio-eventloop-3-3] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-1-1" }, value : TASK_STAGING)
2016-07-27 11:47:28,863 DEBUG [rxnetty-nio-eventloop-3-3] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-1-2" }, value : TASK_STAGING)
2016-07-27 11:47:28,963 DEBUG [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-1-1" }, value : TASK_RUNNING)
2016-07-27 11:47:28,965 DEBUG [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-1-2" }, value : TASK_RUNNING)
2016-07-27 11:47:44,210 DEBUG [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-1-2" }, value : TASK_FINISHED)
2016-07-27 11:47:44,212 DEBUG [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-1-1" }, value : TASK_FINISHED)
2016-07-27 11:47:44,213 INFO  [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.Sleepy  - Launching 2 tasks
2016-07-27 11:47:44,218 DEBUG [rxnetty-nio-eventloop-3-5] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-4-3" }, value : TASK_STAGING)
2016-07-27 11:47:44,219 DEBUG [rxnetty-nio-eventloop-3-5] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-4-4" }, value : TASK_STAGING)
2016-07-27 11:47:44,460 DEBUG [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-4-4" }, value : TASK_RUNNING)
2016-07-27 11:47:44,461 DEBUG [RxComputationScheduler-3 ] c.m.m.r.j.e.framework.sleepy.State   - put(key : { value: "task-4-3" }, value : TASK_RUNNING)

Notice the log messages that set the state to TASK_STAGING are executed on rxnetty-nio-eventloop-3-x, this instead should probably be on RxComputation.

I am converting a Netflix Fenzo-based scheduler to use the HTTP API. One issue I'm encountering is that several of Fenzo's methods accept org.apache.mesos.Protos.Offer. Of course, in my new HTTP-based scheduler, I am receiving org.apache.mesos.v1.Protos.Offer. Is there a clean way to convert from the v1 Protos.Offer to the old Protos.Offer so that I can give it to Fenzo?

Asked about in:

• #75
• #63

Given that the API server can change (see #14), among other situations, what is the simplest way to configure the MesosSchedulerClient that handles most of them? Would it be possible to use a single URL to describe most of the required settings? Investigate the options and implement the most appealing one.

#79 (comment)
Mesos 1.4.0 uses proto3
https://issues.apache.org/jira/browse/MESOS-7228

I think it's time to change dependencies =)
#79 (comment)

In MesosSchedulerClient.openEventStream(), there are some comments regarding backpressure and custom schedulers. Investigate how to improve the code so that the comments can be removed.

Mesos documentation: https://github.com/apache/mesos/blob/master/docs/scheduler-http-api.md#master-detection

Use it as the reference implementation when writing correctness and performance tests for the new streaming API client.

Currently the example framework is using a custom Tuple2 class for representing generic pairs of values. We should really use a library class instead of rolling our own.

I am not completely familiar with the internals of a Mesos cluster, but I have a question. My settings is simple: 1 AWS EC2 instance running mesos and 1 laptop running the Sleepy framework. When I run the framework the MesosClient probes the endpoint using the EC2 public IP address and returns a new link to the master.

2017-03-31 17:58:59,881 INFO [main ] c.m.mesos.rx.java.MesosClient - Probing Mesos server at http://aws-public-ip:5050/redirect

This probe gets redirected to:
Using new Mesos URI: {} http://aws-private-ip:5050/api/v1/scheduler

Because it gets redirected to the private IP and the laptop is on a different subnet it cannot connect to it. In order to get it to work I added a kludge where I look for a system property and if is set then skip the redirect. This kludge works, but I was wondering if there is a better way to do this.

Thanks
Oscar

RxJava 1.x is in feature-freeze with an end of life scheduled March 2018. Is there a timeline for this project to update to RxJava 2.x?

https://github.com/ReactiveX/RxJava/blob/2.x/README.md#version-1x-javadoc

Per offline discussion with @benh this project repo will be moved under the mesosphere org.

• New repo URL: github.com/mesosphere/mesos-rxjava
• New maven groupId: com.mesosphere.mesos.rx.java
• New copyright headers for Mesosphere, Inc
• Keep APLv2 license

As described in the "Master detection" section of the Scheduler HTTP API docs (bottom of page), SUBSCRIBE requests will receive a 307 response status code (Temporary Redirect) if sent to a non-leading Mesos master. The response will include a Location header with the host and port of the leading master.

The MesosSchedulerClient should handle this possibility automatically, for user convenience.

Currently MesosClientScheduler.openEventStream() takes an argument that is expected to be a valid subscribe call to Mesos. There's no validation of the argument though, so any type of call could be passed in, giving unexpected results. Let's figure out the best way to prevent or mitigate invalid calls.

I had a short-term requirement to get my mesos-rxjava-based framework onto DC/OS Enterprise with strict security mode. I would like to submit a PR once tests/documentation are in better shape, and I am simply creating an issue for tracking purposes.

The gist of my change is...

• check if mesosUri is "https". If so, add a "trustAll" SSLEngineFactory to the HTTPClient
• provide a supplier/callback for adding arbitrary headers to the HTTP POST call

I'm not sure how active this project is currently, but I welcome any feedback and suggestions necessary to make the changes acceptable.

PR to follow this week (hopefully)...

I have written a wrapper around this (https://github.com/saiaku-gaming/mesos-scheduler-client) that I use in one of my projects, but now I need to have some extra settings that was introduced in mesos 1.2.0. Are you interested in updating to 1.2.0 or should I fork this project?

Per an offline discussion today, having guava in the project could present a potential problem for other application that have an existing dependency on a specific version of guava.

Time to time I got this exception trying to send ACKNOWLEDGE (not every time):

Caused by: java.lang.IllegalStateException: Content stream is already disposed.
	at io.reactivex.netty.protocol.http.UnicastContentSubject$OnSubscribeAction.call(UnicastContentSubject.java:240)
	at io.reactivex.netty.protocol.http.UnicastContentSubject$OnSubscribeAction.call(UnicastContentSubject.java:213)
	at rx.Observable.unsafeSubscribe(Observable.java:9861)
	at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:48)
	at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:33)
	at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)
	at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)
	at rx.Observable.subscribe(Observable.java:9957)
	at rx.Observable.subscribe(Observable.java:9924)
	at rx.Observable.subscribe(Observable.java:9729)
	at rx.Observable.forEach(Observable.java:6781)
	at com.mesosphere.mesos.rx.java.SinkSubscriber.lambda$onNext$7(SinkSubscriber.java:66)
	at rx.internal.util.ActionSubscriber.onNext(ActionSubscriber.java:39)
	at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:134)
	at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:227)
	... 9 more

To be added once the interface has stabilized.

In my specific scheduler, I use Netflix Fenzo to accumulate (a) Mesos resource offers and (b) Kafka-driven TaskRequests. Fenzo then evaluates the available resources and the needs of my tasks to give back optimal assignments. The issue I'm having with mesos-rxjava is that Fenzo does this asynchronously. I don't have a logical place within the event stream to accept an offer from Mesos and then launch tasks. In other words, in my current, v0, MesosSchedulerDriver scheduler, my decision to launch a task is driven by Fenzo's callback ... not by Mesos.

So with that, to use mesos-rxjava, it seems like I have to just publish/produce my intentions to launch tasks from the Fenzo callback, and then poll/consume those from the rx event stream... perhaps after the heartbeat event.

Three questions:

1. Am I missing something? Is there already a way to asynchrously launch tasks outside the event stream?
2. If there is no way, then is my workaround reasonable? (effectively launching accumulated TaskInfos as a SinkOperation from the heartbeat Event)
3. Do you think that in the future, mesos-rxjava could provide a hook to issue a Call outside the event stream?