OPENDJ-832 (CR-1545) Leverage the work queue for processing requests receiv...

Jean-Noel Rouvignac

15.08.2013 c96af91cdd11f79c11e56d6c7fe33d0edc653ea3

OPENDJ-832 (CR-1545) Leverage the work queue for processing requests received on the HTTP connection handler



Added logging for CONNECT / DISCONNECT operations. All LDAP inner operations are already logged.
Ensured the inner operations are not logged by default.



LDAPClientConnection.java, LDAPConnectionHandler.java:
Moved the disconnect logging from LDAPClientConnection ctor to LDAPConnectionHandler.acceptConnection() in line with other calls to disconnect logging.

HTTPClientConnection.java:
Moved the disconnect logging from HTTPClientConnection ctor to CollectClientConnectionsFilter..doFilter() in line with other calls to disconnect logging.
In sendIntermediateResponseMessage(), throw an exception.
In disconnect(), removed the connection from the connection handler + log the disconnect.

HTTPConnectionHandler.java:
Added addClientConnection() and removeClientConnection().

CollectClientConnectionsFilter.java:
In doFilter(), called HTTPConnectionHandler.addClientConnection() and do not remove the client connection because it is removed when Rest2LDAP calls Connection.close() + called logConnect() + call client disconnect if we sent back the unauthorized status code.

ClientConnection.java
Added isInnerConnection().

Operation.java, AbstractOperation.java, OperationWrapper.java:
Added isInnerOperation() and setInnerOperation().

AbstractTextAccessLogPublisher.java:
Called Operation.isInnerOperation() and ClientConnection.isInnerConnection() instead of checking the connectionID.

AbstractTextAccessLogPublisherTest.java:
Updated the test.

SdkConnectionAdapter.java
In enqueueOperation(), setInnerOperation().
In close(UnbindRequest, String), do not issue run the UnbindOperation if we were not authenticated, but issue a disconnect log message.

 opends/src/server/org/opends/server/api/ClientConnection.java

@@ -23,7 +23,7 @@
 *
 *
 *      Copyright 2006-2009 Sun Microsystems, Inc.
 *      Portions copyright 2011-2012 ForgeRock AS
 *      Portions copyright 2011-2013 ForgeRock AS
 */
package org.opends.server.api;

@@ -109,41 +109,47 @@
   */
  protected AtomicBoolean bindOrStartTLSInProgress;

  // Indicates whether any necessary finalization work has been done
  // for this client connection.
  /**
   *  Indicates whether any necessary finalization work has been done for this
   *  client connection.
   */
  private boolean finalized;

  // The set of privileges assigned to this client connection.
  /** The set of privileges assigned to this client connection. */
  private HashSet<Privilege> privileges;

  // The size limit for use with this client connection.
  /** The size limit for use with this client connection. */
  private int sizeLimit;

  // The time limit for use with this client connection.
  /** The time limit for use with this client connection. */
  private int timeLimit;

  // The lookthrough limit for use with this client connection.
  /** The lookthrough limit for use with this client connection. */
  private int lookthroughLimit;

  // The time that this client connection was established.
  /** The time that this client connection was established. */
  private final long connectTime;

  // The idle time limit for this client connection.
  /** The idle time limit for this client connection. */
  private long idleTimeLimit;

  // The opaque information used for storing intermediate state
  // information needed across multi-stage SASL binds.
  /**
   * The opaque information used for storing intermediate state information
   * needed across multi-stage SASL binds.
   */
  private Object saslAuthState;

  // A string representation of the time that this client connection
  // was established.
  /**
   * A string representation of the time that this client connection was
   * established.
   */
  private final String connectTimeString;

  // A set of persistent searches registered for this client.
  /** A set of persistent searches registered for this client. */
  private final CopyOnWriteArrayList<PersistentSearch>
      persistentSearches;

  // The network group to which the connection belongs to.
  /** The network group to which the connection belongs to. */
  private NetworkGroup networkGroup;

  /** Need to evaluate the network group for the first operation. */
@@ -1767,5 +1773,17 @@
  {
    saslBindInProgress.set(false);
  }

  /**
   * Returns whether this connection is used for inner work not directly
   * requested by an external client.
   *
   * @return {@code true} if this is an inner connection, {@code false}
   *         otherwise
   */
  public boolean isInnerConnection()
  {
    return getConnectionID() < 0;
}

}

 opends/src/server/org/opends/server/core/OperationWrapper.java

@@ -33,8 +33,8 @@
import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
import org.opends.server.api.ClientConnection;
import org.opends.server.types.*;
import org.opends.server.controls.ControlDecoder;
import org.opends.server.types.*;


/**
@@ -355,6 +355,15 @@
   * {@inheritDoc}
   */
  @Override
  public boolean isInnerOperation()
  {
    return operation.isInnerOperation();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isInternalOperation()
  {
    return operation.isInternalOperation();
@@ -454,6 +463,15 @@
   * {@inheritDoc}
   */
  @Override
  public void setInnerOperation(boolean isInnerOperation)
  {
    operation.setInnerOperation(isInnerOperation);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void setInternalOperation(boolean isInternalOperation)
  {
    operation.setInternalOperation(isInternalOperation);

 opends/src/server/org/opends/server/loggers/AbstractTextAccessLogPublisher.java

@@ -962,8 +962,7 @@
    @Override
    public boolean isConnectLoggable(final ClientConnection connection)
    {
      final long connectionID = connection.getConnectionID();
      if (connectionID >= 0 || !suppressInternalOperations)
      if (!connection.isInnerConnection() || !suppressInternalOperations)
      {
        switch (policy)
        {
@@ -989,8 +988,7 @@
    @Override
    public boolean isDisconnectLoggable(final ClientConnection connection)
    {
      final long connectionID = connection.getConnectionID();
      if (connectionID >= 0 || !suppressInternalOperations)
      if (!connection.isInnerConnection() || !suppressInternalOperations)
      {
        switch (policy)
        {
@@ -1071,21 +1069,9 @@
     */
    boolean isLoggable(final Operation operation)
    {
      final long connectionID = operation.getConnectionID();
      if (connectionID < 0)
      {
        // This is an internal operation.
        if (operation.isSynchronizationOperation())
        {
          return !suppressSynchronizationOperations;
        }
        else
        {
          return !suppressInternalOperations;
        }
      }

      return true;
      return !((suppressInternalOperations && operation.isInnerOperation())
          || (suppressSynchronizationOperations
              && operation.isSynchronizationOperation()));
    }
  }


 opends/src/server/org/opends/server/protocols/http/CollectClientConnectionsFilter.java

@@ -28,6 +28,7 @@

import static org.forgerock.opendj.adapter.server2x.Converters.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;
import static org.opends.server.loggers.ErrorLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.util.StaticUtils.*;
@@ -37,7 +38,6 @@
import java.net.UnknownHostException;
import java.text.ParseException;
import java.util.Collection;
import java.util.Map;

import javax.servlet.FilterChain;
import javax.servlet.FilterConfig;
@@ -61,7 +61,6 @@
import org.forgerock.opendj.rest2ldap.servlet.Rest2LDAPContextFactory;
import org.opends.messages.Message;
import org.opends.server.admin.std.server.ConnectionHandlerCfg;
import org.opends.server.api.ClientConnection;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.schema.SchemaConstants;
import org.opends.server.types.AddressMask;
@@ -120,17 +119,18 @@
  public void doFilter(ServletRequest request, ServletResponse response,
      FilterChain chain)
  {
    final Map<ClientConnection, ClientConnection> clientConnections =
        this.connectionHandler.getClientConnectionsMap();
    final HTTPClientConnection clientConnection =
        new HTTPClientConnection(this.connectionHandler, request);
    clientConnections.put(clientConnection, clientConnection);
    this.connectionHandler.addClientConnection(clientConnection);
    try
    {
      if (!canProcessRequest(request, clientConnection))
      {
        return;
      }
      // logs the connect after all the possible disconnect reasons have been
      // checked.
      logConnect(clientConnection);

      Connection connection = new SdkConnectionAdapter(clientConnection);

@@ -157,6 +157,8 @@
      // The user could not be authenticated. Send an HTTP Basic authentication
      // challenge if HTTP Basic authentication is enabled.
      sendUnauthorizedResponseWithHTTPBasicAuthChallenge(response);
      clientConnection.disconnect(DisconnectReason.INVALID_CREDENTIALS, false,
          null);
    }
    catch (Exception e)
    {
@@ -174,10 +176,6 @@
      clientConnection
          .disconnect(DisconnectReason.SERVER_ERROR, false, message);
    }
    finally
    {
      clientConnections.remove(clientConnection);
    }
  }

  private boolean canProcessRequest(ServletRequest request,
@@ -190,7 +188,8 @@
    // established).
    if (clientConnection.getConnectionID() < 0)
    {
      // The connection will have already been closed.
      clientConnection.disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
      return false;
    }


 opends/src/server/org/opends/server/protocols/http/HTTPClientConnection.java

@@ -28,6 +28,7 @@

import static org.forgerock.opendj.adapter.server2x.Converters.*;
import static org.opends.messages.ProtocolMessages.*;
import static org.opends.server.loggers.AccessLogger.*;
import static org.opends.server.loggers.debug.DebugLogger.*;

import java.net.InetAddress;
@@ -47,7 +48,6 @@
import org.opends.messages.Message;
import org.opends.messages.MessageBuilder;
import org.opends.server.api.ClientConnection;
import org.opends.server.api.ConnectionHandler;
import org.opends.server.core.DirectoryServer;
import org.opends.server.core.SearchOperation;
import org.opends.server.loggers.debug.DebugTracer;
@@ -171,11 +171,6 @@
    this.request = request;

    this.connectionID = DirectoryServer.newConnectionAccepted(this);
    if (this.connectionID < 0)
    {
      disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
    }
  }

  /** {@inheritDoc} */
@@ -187,7 +182,7 @@

  /** {@inheritDoc} */
  @Override
  public ConnectionHandler<?> getConnectionHandler()
  public HTTPConnectionHandler getConnectionHandler()
  {
    return connectionHandler;
  }
@@ -315,8 +310,7 @@
  protected boolean sendIntermediateResponseMessage(
      IntermediateResponse intermediateResponse)
  {
    // TODO Auto-generated method stub
    return false;
    throw new RuntimeException("Not implemented");
  }

  /**
@@ -372,6 +366,10 @@
          .getClosureMessage()));
    }
    finalizeConnectionInternal();


    this.connectionHandler.removeClientConnection(this);
    logDisconnect(this, disconnectReason, message);
  }

  /** {@inheritDoc} */
@@ -475,7 +473,7 @@
            op.operation.abort(cancelRequest);
          }
          catch (Exception e)
          { // make sure all operations are cancelled, no mattter what
          { // make sure all operations are cancelled, no matter what
            if (debugEnabled())
            {
              TRACER.debugCaught(DebugLogLevel.ERROR, e);
@@ -593,4 +591,11 @@
  {
    return connectionValid;
  }

  /** {@inheritDoc} */
  @Override
  public boolean isInnerConnection()
  {
    return true;
  }
}

 opends/src/server/org/opends/server/protocols/http/HTTPConnectionHandler.java

@@ -200,6 +200,17 @@
        && !this.currentConfig.isAuthenticationRequired();
  }

  /**
   * Registers a client connection to track it.
   *
   * @param clientConnection
   *          the client connection to register
   */
  void addClientConnection(ClientConnection clientConnection)
  {
    clientConnections.put(clientConnection, clientConnection);
  }

  /** {@inheritDoc} */
  @Override
  public ConfigChangeResult applyConfigurationChange(
@@ -347,16 +358,6 @@
    return clientConnections.keySet();
  }

  /**
   * Gives access to the clientConnections to classes in this package.
   *
   * @return the Map containing the current client connections
   */
  Map<ClientConnection, ClientConnection> getClientConnectionsMap()
  {
    return clientConnections;
  }

  /** {@inheritDoc} */
  @Override
  public DN getComponentEntryDN()
@@ -644,6 +645,17 @@
    }
  }

  /**
   * Unregisters a client connection to stop tracking it.
   *
   * @param clientConnection
   *          the client connection to unregister
   */
  void removeClientConnection(ClientConnection clientConnection)
  {
    clientConnections.remove(clientConnection);
  }

  /** {@inheritDoc} */
  @Override
  public void run()

 opends/src/server/org/opends/server/protocols/http/SdkConnectionAdapter.java

@@ -69,8 +69,10 @@
import org.opends.server.core.UnbindOperationBasis;
import org.opends.server.core.WorkQueueStrategy;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.types.AuthenticationInfo;
import org.opends.server.types.ByteString;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DisconnectReason;
import org.opends.server.types.Operation;

import com.forgerock.opendj.util.AsynchronousFutureResult;
@@ -125,6 +127,8 @@

    try
    {
      operation.setInnerOperation(this.clientConnection.isInnerConnection());

      // need this raw cast here to fool the compiler's generic type safety
      // Problem here is due to the generic type R on enqueueOperation()
      clientConnection.addOperationInProgress(operation,
@@ -201,6 +205,9 @@
  @Override
  public void close(UnbindRequest request, String reason)
  {
    AuthenticationInfo authInfo = this.clientConnection.getAuthenticationInfo();
    if (authInfo != null && authInfo.isAuthenticated())
    {
    final int messageID = nextMessageID.get();
    UnbindOperationBasis operation =
        new UnbindOperationBasis(clientConnection, messageID, messageID,
@@ -208,7 +215,11 @@

    // run synchronous
    operation.run();

    }
    else
    {
      this.clientConnection.disconnect(DisconnectReason.UNBIND, false, null);
    }
    isClosed = true;
  }


 opends/src/server/org/opends/server/protocols/ldap/LDAPClientConnection.java

@@ -519,11 +519,6 @@
    }

    connectionID = DirectoryServer.newConnectionAccepted(this);
    if (connectionID < 0)
    {
      disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
    }
  }

  /**

 opends/src/server/org/opends/server/protocols/ldap/LDAPConnectionHandler.java

@@ -1239,7 +1239,8 @@
        clientChannel, getProtocol());
    if (clientConnection.getConnectionID() < 0)
    {
      // The connection will have already been closed.
      clientConnection.disconnect(DisconnectReason.ADMIN_LIMIT_EXCEEDED, true,
          ERR_CONNHANDLER_REJECTED_BY_SERVER.get());
      return;
    }


 opends/src/server/org/opends/server/types/AbstractOperation.java

@@ -116,6 +116,7 @@
   * itself rather than requested by an external client.
   */
  private boolean isInternalOperation;
  private Boolean isInnerOperation;

  /**
   * Indicates whether this operation is involved in data synchronization
@@ -621,65 +622,47 @@



  /**
   * Indicates whether this is an internal operation rather than one
   * that was requested by an external client.
   *
   * @return  {@code true} if this is an internal operation, or
   *          {@code false} if it is not.
   */
  /** {@inheritDoc} */
  @Override
  public final boolean isInternalOperation()
  {
    return isInternalOperation;
  }



  /**
   * Specifies whether this is an internal operation rather than one
   * that was requested by an external client.  This may not be called
   * from within a plugin.
   *
   * @param  isInternalOperation  Specifies whether this is an
   *                              internal operation rather than one
   *                              that was requested by an external
   *                              client.
   */
  /** {@inheritDoc} */
  @Override
  public final void setInternalOperation(boolean isInternalOperation)
  {
    this.isInternalOperation = isInternalOperation;
  }

  /** {@inheritDoc} */
  @Override
  public boolean isInnerOperation()
  {
    if (this.isInnerOperation != null)
    {
      return this.isInnerOperation;
    }
    return isInternalOperation();
  }

  /** {@inheritDoc} */
  @Override
  public void setInnerOperation(boolean isInnerOperation)
  {
    this.isInnerOperation = isInnerOperation;
  }


  /**
   * Indicates whether this is a synchronization operation rather than
   * one that was requested by an external client.
   *
   * @return  {@code true} if this is a data synchronization
   *          operation, or {@code false} if it is not.
   */
  /** {@inheritDoc} */
  @Override
  public final boolean isSynchronizationOperation()
  {
    return isSynchronizationOperation;
  }



  /**
   * Specifies whether this is a synchronization operation rather than
   * one that was requested by an external client.  This method may
   * not be called from within a plugin.
   *
   * @param  isSynchronizationOperation  Specifies whether this is a
   *                                     synchronization operation
   *                                     rather than one that was
   *                                     requested by an external
   *                                     client.
   */
  /** {@inheritDoc} */
  @Override
  public final void setSynchronizationOperation(
                         boolean isSynchronizationOperation)

 opends/src/server/org/opends/server/types/Operation.java

@@ -367,6 +367,31 @@
      isInternalOperation);

  /**
   * Indicates whether this is an inner operation rather than one that was
   * directly requested by an external client. Said otherwise, inner operations
   * include internal operations, but also operations in the server indirectly
   * mandated by external requests like Rest2LDAP for example. This may not be
   * called from within a plugin.
   *
   * @return {@code true} if this is an inner operation, or {@code false} if it
   *         is not.
   */
  boolean isInnerOperation();

  /**
   * Specifies whether this is an inner operation rather than one that was
   * directly requested by an external client. Said otherwise, inner operations
   * include internal operations, but also operations in the server indirectly
   * mandated by external requests like Rest2LDAP for example. This may not be
   * called from within a plugin.
   *
   * @param isInnerOperation
   *          Specifies whether this is an inner operation rather than one that
   *          was requested by an external client.
   */
  void setInnerOperation(boolean isInnerOperation);

  /**
   * Indicates whether this is a synchronization operation rather than
   * one that was requested by an external client.
   *

 opends/tests/unit-tests-testng/src/server/org/opends/server/loggers/AbstractTextAccessLogPublisherTest.java

@@ -42,40 +42,42 @@
  @DataProvider(name = "isLoggableData")
  public Object[][] getIsLoggableData()
  {
    // when suppress is set to true and the corresponding operation is set to
    // true too, then the operation is not loggable.
    // You can read the array like this: read two by two from line start, if
    // both are true in a pair, then the expected result is false (not loggable)
    return new Object[][] {
      { 1L, false, false, false, false, true },
      { -1L, true, true, true, true, false },
      { -1L, true, true, true, false, false },
      { -1L, true, true, false, true, false },
      { -1L, true, true, false, false, false },
      { -1L, true, false, true, true, false },
      { -1L, true, false, true, false, false },// this will change
      { -1L, true, false, false, true, true },
      { -1L, true, false, false, false, true },
      { -1L, false, true, true, true, true },// this will change
      { -1L, false, true, true, false, true },
      { -1L, false, true, false, true, true },
      { -1L, false, true, false, false, true },
      { -1L, false, false, true, true, false },
      { -1L, false, false, true, false, false },// this will change
      { -1L, false, false, false, true, true },
      { -1L, false, false, false, false, true }, };
      { true, true, true, true, false },
      { true, true, true, false, false },
      { true, true, false, true, false },
      { true, true, false, false, false },
      { true, false, true, true, false },
      { true, false, true, false, true },
      { true, false, false, true, true },
      { true, false, false, false, true },
      { false, true, true, true, false },
      { false, true, true, false, true },
      { false, true, false, true, true },
      { false, true, false, false, true },
      { false, false, true, true, false },
      { false, false, true, false, true },
      { false, false, false, true, true },
      { false, false, false, false, true }, };
  }

  @Test(dataProvider = "isLoggableData")
  public void rootFilterIsLoggable(long connectionID,
      boolean suppressSynchronization, boolean isSynchronizationOp,
      boolean suppressInternal, boolean isInternalOp, boolean testResult)
  public void rootFilterIsLoggable(boolean suppressSynchronization,
      boolean isSynchronizationOp, boolean suppressInternal,
      boolean isInternalOp, boolean expectedTestResult)
  {
    final Operation operation = mock(Operation.class);
    when(operation.getConnectionID()).thenReturn(connectionID);
    when(operation.isSynchronizationOperation())
        .thenReturn(isSynchronizationOp);
    when(operation.isInternalOperation()).thenReturn(isInternalOp);
    when(operation.isInnerOperation()).thenReturn(isInternalOp);

    final RootFilter filter =
        new RootFilter(suppressInternal, suppressSynchronization, null, null);
    assertEquals(filter.isLoggable(operation), testResult);
    assertEquals(filter.isLoggable(operation), expectedTestResult);
  }

}