Move logic for creating a session into the LocalDistributor

We originally lifted this logic up to the `Distributor` class when
working on a version of the Distributor that stored state in either a
database or redis. However, we never implemented those, and there is
restructuring work to do. Moving the logic back into the main
implementation makes this easier.

Author: shs96c

java/server/src/org/openqa/selenium/grid/distributor/CreateSession.java

@@ -0,0 +1,63 @@
+// Licensed to the Software Freedom Conservancy (SFC) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The SFC 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.openqa.selenium.grid.distributor;
+
+import org.openqa.selenium.SessionNotCreatedException;
+import org.openqa.selenium.grid.data.CreateSessionResponse;
+import org.openqa.selenium.grid.data.SessionRequest;
+import org.openqa.selenium.internal.Either;
+import org.openqa.selenium.internal.Require;
+import org.openqa.selenium.remote.http.Contents;
+import org.openqa.selenium.remote.http.HttpHandler;
+import org.openqa.selenium.remote.http.HttpRequest;
+import org.openqa.selenium.remote.http.HttpResponse;
+
+import java.io.UncheckedIOException;
+import java.util.Map;
+
+import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
+import static java.util.Collections.singletonMap;
+
+class CreateSession implements HttpHandler {
+
+  private final Distributor distributor;
+
+  CreateSession(Distributor distributor) {
+    this.distributor = Require.nonNull("Distributor", distributor);
+  }
+
+  @Override
+  public HttpResponse execute(HttpRequest req) throws UncheckedIOException {
+    SessionRequest request = Contents.fromJson(req, SessionRequest.class);
+
+    Either<SessionNotCreatedException, CreateSessionResponse> result = distributor.newSession(request);
+
+    HttpResponse res = new HttpResponse();
+    Map<String, Object> value;
+    if (result.isLeft()) {
+      res.setStatus(HTTP_INTERNAL_ERROR);
+      value = singletonMap("value", result.left());
+    } else {
+      value = singletonMap("value", result.right());
+    }
+
+    res.setContent(Contents.asJson(value));
+
+    return res;
+  }
+}

java/server/src/org/openqa/selenium/grid/distributor/Distributor.java

@@ -17,63 +17,42 @@
 
 package org.openqa.selenium.grid.distributor;
 
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import org.openqa.selenium.Capabilities;
-import org.openqa.selenium.RetrySessionRequestException;
 import org.openqa.selenium.SessionNotCreatedException;
-import org.openqa.selenium.grid.data.CreateSessionRequest;
 import org.openqa.selenium.grid.data.CreateSessionResponse;
 import org.openqa.selenium.grid.data.DistributorStatus;
 import org.openqa.selenium.grid.data.NodeId;
 import org.openqa.selenium.grid.data.NodeStatus;
 import org.openqa.selenium.grid.data.Session;
-import org.openqa.selenium.grid.data.SlotId;
+import org.openqa.selenium.grid.data.SessionRequest;
 import org.openqa.selenium.grid.distributor.selector.SlotSelector;
 import org.openqa.selenium.grid.node.Node;
 import org.openqa.selenium.grid.security.RequiresSecretFilter;
 import org.openqa.selenium.grid.security.Secret;
 import org.openqa.selenium.grid.sessionmap.SessionMap;
-import org.openqa.selenium.grid.data.SessionRequest;
 import org.openqa.selenium.internal.Either;
 import org.openqa.selenium.internal.Require;
 import org.openqa.selenium.json.Json;
-import org.openqa.selenium.remote.SessionId;
 import org.openqa.selenium.remote.http.HttpClient;
 import org.openqa.selenium.remote.http.HttpRequest;
 import org.openqa.selenium.remote.http.HttpResponse;
 import org.openqa.selenium.remote.http.Routable;
 import org.openqa.selenium.remote.http.Route;
-import org.openqa.selenium.remote.tracing.AttributeKey;
-import org.openqa.selenium.remote.tracing.EventAttribute;
-import org.openqa.selenium.remote.tracing.EventAttributeValue;
-import org.openqa.selenium.remote.tracing.Span;
 import org.openqa.selenium.remote.tracing.SpanDecorator;
-import org.openqa.selenium.remote.tracing.Status;
 import org.openqa.selenium.remote.tracing.Tracer;
 import org.openqa.selenium.status.HasReadyState;
 
 import java.io.UncheckedIOException;
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Predicate;
 import java.util.logging.Logger;
-import java.util.stream.Collectors;
 
-import static org.openqa.selenium.remote.RemoteTags.CAPABILITIES;
-import static org.openqa.selenium.remote.RemoteTags.CAPABILITIES_EVENT;
-import static org.openqa.selenium.remote.RemoteTags.SESSION_ID;
-import static org.openqa.selenium.remote.RemoteTags.SESSION_ID_EVENT;
 import static org.openqa.selenium.remote.http.Route.delete;
 import static org.openqa.selenium.remote.http.Route.get;
 import static org.openqa.selenium.remote.http.Route.post;
-import static org.openqa.selenium.remote.tracing.Tags.EXCEPTION;
 
 /**
  * Responsible for being the central place where the {@link Node}s
@@ -117,20 +96,14 @@ public abstract class Distributor implements HasReadyState, Predicate<HttpReques
 
   private final Route routes;
   protected final Tracer tracer;
-  private final SlotSelector slotSelector;
-  private final SessionMap sessions;
   private final ReadWriteLock lock = new ReentrantReadWriteLock(true);
 
   protected Distributor(
     Tracer tracer,
     HttpClient.Factory httpClientFactory,
-    SlotSelector slotSelector,
-    SessionMap sessions,
     Secret registrationSecret) {
     this.tracer = Require.nonNull("Tracer", tracer);
     Require.nonNull("HTTP client factory", httpClientFactory);
-    this.slotSelector = Require.nonNull("Slot selector", slotSelector);
-    this.sessions = Require.nonNull("Session map", sessions);
 
     Require.nonNull("Registration secret", registrationSecret);
 
@@ -150,132 +123,16 @@ protected Distributor(
           .to(params ->
                 new RemoveNode(this, new NodeId(UUID.fromString(params.get("nodeId")))))
           .with(requiresSecret),
-      get("/se/grid/distributor/status")
+      post("/se/grid/distributor/session")
+        .to(() -> new CreateSession(this))
+        .with(requiresSecret),
+    get("/se/grid/distributor/status")
           .to(() -> new GetDistributorStatus(this))
           .with(new SpanDecorator(tracer, req -> "distributor.status")));
   }
 
-  public Either<SessionNotCreatedException, CreateSessionResponse> newSession(SessionRequest request)
-    throws SessionNotCreatedException {
-    Require.nonNull("Requests to process", request);
-
-    Span span = tracer.getCurrentContext().createSpan("distributor.create_session_response");
-    Map<String, EventAttributeValue> attributeMap = new HashMap<>();
-    try {
-      attributeMap.put(AttributeKey.LOGGER_CLASS.getKey(),
-        EventAttribute.setValue(getClass().getName()));
-
-      Iterator<Capabilities> iterator = request.getDesiredCapabilities().iterator();
-      attributeMap.put("request.payload", EventAttribute.setValue(request.getDesiredCapabilities().toString()));
-      String sessionReceivedMessage = "Session request received by the distributor";
-      span.addEvent(sessionReceivedMessage, attributeMap);
-      LOG.info(String.format("%s: \n %s", sessionReceivedMessage, request.getDesiredCapabilities()));
-
-      if (!iterator.hasNext()) {
-        SessionNotCreatedException exception =
-          new SessionNotCreatedException("No capabilities found in session request payload");
-        EXCEPTION.accept(attributeMap, exception);
-        attributeMap.put(AttributeKey.EXCEPTION_MESSAGE.getKey(),
-          EventAttribute.setValue("Unable to create session. No capabilities found: " +
-            exception.getMessage()));
-        span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
-        return Either.left(exception);
-      }
-
-      Either<SessionNotCreatedException, CreateSessionResponse> selected;
-      CreateSessionRequest firstRequest = new CreateSessionRequest(
-        request.getDownstreamDialects(),
-        iterator.next(),
-        ImmutableMap.of("span", span));
-
-      Lock writeLock = this.lock.writeLock();
-      writeLock.lock();
-      try {
-        Set<NodeStatus> model = ImmutableSet.copyOf(getAvailableNodes());
-
-        // Reject new session immediately if no node has the required capabilities
-        boolean hostsWithCaps = model.stream()
-          .anyMatch(nodeStatus -> nodeStatus.hasCapability(firstRequest.getDesiredCapabilities()));
-
-        if (!hostsWithCaps) {
-          String errorMessage = String.format(
-            "No Node supports the required capabilities: %s",
-            request.getDesiredCapabilities().stream().map(Capabilities::toString)
-              .collect(Collectors.joining(", ")));
-          SessionNotCreatedException exception = new SessionNotCreatedException(errorMessage);
-          span.setAttribute(AttributeKey.ERROR.getKey(), true);
-          span.setStatus(Status.ABORTED);
-
-          EXCEPTION.accept(attributeMap, exception);
-          attributeMap.put(AttributeKey.EXCEPTION_MESSAGE.getKey(),
-            EventAttribute.setValue("Unable to create session: " + exception.getMessage()));
-          span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
-          return Either.left(exception);
-        }
-
-        // Find a Node that supports the capabilities present in the new session
-        Set<SlotId> slotIds = slotSelector.selectSlot(firstRequest.getDesiredCapabilities(), model);
-        if (!slotIds.isEmpty()) {
-          selected = reserve(slotIds.iterator().next(), firstRequest);
-        } else {
-          String errorMessage =
-            String.format(
-              "Unable to find provider for session: %s",
-              request.getDesiredCapabilities().stream().map(Capabilities::toString)
-                .collect(Collectors.joining(", ")));
-          SessionNotCreatedException exception = new RetrySessionRequestException(errorMessage);
-          selected = Either.left(exception);
-        }
-      } finally {
-        writeLock.unlock();
-      }
-
-      if (selected.isRight()) {
-        CreateSessionResponse sessionResponse = selected.right();
-
-        sessions.add(sessionResponse.getSession());
-        SessionId sessionId = sessionResponse.getSession().getId();
-        Capabilities caps = sessionResponse.getSession().getCapabilities();
-        String sessionUri = sessionResponse.getSession().getUri().toString();
-        SESSION_ID.accept(span, sessionId);
-        CAPABILITIES.accept(span, caps);
-        SESSION_ID_EVENT.accept(attributeMap, sessionId);
-        CAPABILITIES_EVENT.accept(attributeMap, caps);
-        span.setAttribute(AttributeKey.SESSION_URI.getKey(), sessionUri);
-        attributeMap.put(AttributeKey.SESSION_URI.getKey(), EventAttribute.setValue(sessionUri));
-
-        String sessionCreatedMessage = "Session created by the distributor";
-        span.addEvent(sessionCreatedMessage, attributeMap);
-        LOG.info(String.format("%s. Id: %s, Caps: %s", sessionCreatedMessage, sessionId, caps));
-        return Either.right(sessionResponse);
-
-      } else {
-        return selected;
-      }
-    } catch (SessionNotCreatedException e) {
-      span.setAttribute(AttributeKey.ERROR.getKey(), true);
-      span.setStatus(Status.ABORTED);
-
-      EXCEPTION.accept(attributeMap, e);
-      attributeMap.put(AttributeKey.EXCEPTION_MESSAGE.getKey(),
-        EventAttribute.setValue("Unable to create session: " + e.getMessage()));
-      span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
-
-      return Either.left(e);
-    } catch (UncheckedIOException e) {
-      span.setAttribute(AttributeKey.ERROR.getKey(), true);
-      span.setStatus(Status.UNKNOWN);
-
-      EXCEPTION.accept(attributeMap, e);
-      attributeMap.put(AttributeKey.EXCEPTION_MESSAGE.getKey(),
-        EventAttribute.setValue("Unknown error in LocalDistributor while creating session: " + e.getMessage()));
-      span.addEvent(AttributeKey.EXCEPTION_EVENT.getKey(), attributeMap);
-
-      return Either.left(new SessionNotCreatedException(e.getMessage(), e));
-    } finally {
-      span.close();
-    }
-  }
+  public abstract Either<SessionNotCreatedException, CreateSessionResponse> newSession(SessionRequest request)
+    throws SessionNotCreatedException;
 
   public abstract Distributor add(Node node);
 
@@ -285,12 +142,6 @@ public Either<SessionNotCreatedException, CreateSessionResponse> newSession(Sess
 
   public abstract DistributorStatus getStatus();
 
-  protected abstract Set<NodeStatus> getAvailableNodes();
-
-  protected abstract Either<SessionNotCreatedException, CreateSessionResponse> reserve(
-    SlotId slot,
-    CreateSessionRequest request);
-
   @Override
   public boolean test(HttpRequest httpRequest) {
     return matches(httpRequest);