Cluster + Security

-----------------------------------

* initial observation of a problem was a 2% failure rate in perftests
  of 20,000 messages against a cluster with security enabled.
  Problem was occasional receit of encrypted frames before the
  security codec had been enabled.  This is fixed with locking in
  cluster code (no new locks in broker code) and a callback that is
  fired by broker::ConnectionHandler::Handler to tell the cluster
  code when the opening handshake has finished.
  This was never a problem in the non-clustered broker before because
  everything happened in a single thread.

* the brokers that "shadow" the connection must not have null
  authenticators rather than real ones, so that they go through all
  the motions but don't do anythig.  Only the directly-connected
  broker can perform the security handshake.

* once the directly-connected broker receives the real user ID
  from its callback, it mcasts that ID to all other brokers.
  Otherwise the shadowing brokers will al think that the user ID
  is "anonymous".
  Check this by doing a substantial perftest, and using
      qpid-stat -c localhost:PORT
  to confirm that the brokers all have the same userID for the
  same connection.

* the user ID, negotiated during the Sasl security startup, is
   communicated from the directly connected broker to all other
   cluster brokers.

* If security is *not* being used, then this code should *not* tell
  the brokers anything about the userID -- or it will step on the value
  that is being set by other code pathways.

* test program at cpp/src/tests/cluster_authentication_soak is not yet
  fully automated -- run it with something like
  "sudo ./cluster_authentication_soak 500"



git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@944158 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 255 insertions, 3 deletions

diff --git a/cpp/src/tests/ForkedBroker.cpp b/cpp/src/tests/ForkedBroker.cpp
index 7c81d303fc..529774df98 100644
--- a/cpp/src/tests/ForkedBroker.cpp
+++ b/cpp/src/tests/ForkedBroker.cpp
@@ -139,6 +139,7 @@ void ForkedBroker::init(const Args& userArgs) {
         std::transform(args.begin(), args.end(), argv.begin(), boost::bind(&std::string::c_str, _1));
         argv.push_back(0);
         QPID_LOG(debug, "ForkedBroker exec " << prog << ": " << args);
+
         execv(prog, const_cast<char* const*>(&argv[0]));
         QPID_LOG(critical, "execv failed to start broker: prog=\"" << prog << "\"; args=\"" << args << "\"; errno=" << errno << " (" << std::strerror(errno) << ")");
         ::exit(1);
diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am
index 6133fc2e49..02b006665e 100644
--- a/cpp/src/tests/Makefile.am
+++ b/cpp/src/tests/Makefile.am
@@ -245,6 +245,11 @@ failover_soak_INCLUDES=$(PUBLIC_INCLUDES)
 failover_soak_SOURCES=failover_soak.cpp  ForkedBroker.h ForkedBroker.cpp
 failover_soak_LDADD=$(lib_client) $(lib_broker)
 
+check_PROGRAMS+=cluster_authentication_soak
+cluster_authentication_soak_INCLUDES=$(PUBLIC_INCLUDES)
+cluster_authentication_soak_SOURCES=cluster_authentication_soak.cpp  ForkedBroker.h ForkedBroker.cpp
+cluster_authentication_soak_LDADD=$(lib_client) $(lib_broker)
+
 check_PROGRAMS+=declare_queues
 declare_queues_INCLUDES=$(PUBLIC_INCLUDES)
 declare_queues_SOURCES=declare_queues.cpp  
@@ -355,7 +360,7 @@ CLEANFILES+=valgrind.out *.log *.vglog* dummy_test qpidd.port $(unit_wrappers)
 # Not run under valgrind, too slow
 
 LONG_TESTS+=start_broker fanout_perftest shared_perftest multiq_perftest topic_perftest run_ring_queue_test stop_broker \
- run_failover_soak reliable_replication_test \
+ run_failover_soak run_cluster_authentication_soak reliable_replication_test \
  federated_cluster_test_with_node_failure
 
 EXTRA_DIST+=						\
@@ -364,6 +369,7 @@ EXTRA_DIST+=						\
 	multiq_perftest					\
 	topic_perftest					\
 	run_failover_soak				\
+	run_cluster_authentication_soak                 \
 	reliable_replication_test			\
 	federated_cluster_test_with_node_failure
 
diff --git a/cpp/src/tests/cluster.cmake b/cpp/src/tests/cluster.cmake
index 9084bfb85b..5f7a811007 100644
--- a/cpp/src/tests/cluster.cmake
+++ b/cpp/src/tests/cluster.cmake
@@ -21,8 +21,8 @@
 # Cluster tests cmake fragment, to be included in CMakeLists.txt
 # 
 
-add_executable (failover_soak failover_soak.cpp ForkedBroker.cpp ${platform_test_additions})
-target_link_libraries (failover_soak qpidclient)
+add_executable (failover_soak failover_soak.cpp cluster_authentication_soak cluster_authentication_soak.cpp ForkedBroker.cpp ${platform_test_additions})
+target_link_libraries (failover_soak cluster_authentication_soak qpidclient)
 remember_location(failover_soak)
 
 set (cluster_test_SOURCES
diff --git a/cpp/src/tests/cluster_authentication_soak.cpp b/cpp/src/tests/cluster_authentication_soak.cpp
new file mode 100644
index 0000000000..985c3aa52a
--- /dev/null
+++ b/cpp/src/tests/cluster_authentication_soak.cpp
@@ -0,0 +1,244 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <signal.h>
+#include <fcntl.h>
+
+#include <sys/wait.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/time.h>
+
+#include <string>
+#include <iostream>
+#include <sstream>
+#include <vector>
+
+#include <boost/assign.hpp>
+
+#include "qpid/framing/Uuid.h"
+
+#include <ForkedBroker.h>
+#include <qpid/client/Connection.h>
+
+
+
+
+
+using namespace std;
+using boost::assign::list_of;
+using namespace qpid::framing;
+using namespace qpid::client;
+
+
+namespace qpid {
+namespace tests {
+
+vector<pid_t> brokerPids;
+
+typedef vector<ForkedBroker *> brokerVector;
+
+
+
+
+
+int newbiePort    = 0;
+
+
+
+
+void
+startBroker ( brokerVector & brokers ,
+              int brokerNumber ) {
+    stringstream portSS, prefix;
+    prefix << "soak-" << brokerNumber;
+    std::vector<std::string> argv;
+
+    argv.push_back ("../qpidd");
+    argv.push_back ("--no-module-dir");
+    argv.push_back ("--load-module=../.libs/cluster.so");
+    argv.push_back ("--cluster-name=micks_test_cluster");
+    argv.push_back ("--cluster-username=guest");
+    argv.push_back ("--cluster-password=guest");
+    argv.push_back ("--cluster-mechanism=ANONYMOUS");
+    argv.push_back ("TMP_DATA_DIR");
+    argv.push_back ("--auth=yes");
+    argv.push_back ("--mgmt-enable=yes");
+    argv.push_back ("--log-prefix");
+    argv.push_back (prefix.str());
+    argv.push_back ("--log-to-file");
+    argv.push_back (prefix.str()+".log");
+
+    ForkedBroker * newbie = new ForkedBroker (argv);
+    newbiePort = newbie->getPort();
+    brokers.push_back ( newbie );
+}
+
+
+
+
+bool
+runPerftest ( ) {
+    stringstream portSs;
+    portSs << newbiePort;
+
+    char const *  path = "./perftest";
+
+    vector<char const *> argv;
+    argv.push_back ( "./perftest" );
+    argv.push_back ( "-p" );
+    argv.push_back ( portSs.str().c_str() );
+    argv.push_back ( "--username" );
+    argv.push_back ( "guest" );
+    argv.push_back ( "--password" );
+    argv.push_back ( "guest" );
+    argv.push_back ( "--mechanism" );
+    argv.push_back ( "DIGEST-MD5" );
+    argv.push_back ( "--count" );
+    argv.push_back ( "20000" );
+    argv.push_back ( 0 );
+
+    pid_t pid = fork();
+
+    if ( ! pid ) {
+        int i=open("/dev/null",O_RDWR);
+        dup2 ( i, fileno(stdout) );
+        dup2 ( i, fileno(stderr) );
+
+        execv ( path, const_cast<char * const *>(&argv[0]) );
+        // The exec failed: we are still in parent process.
+        perror ( "error running perftest: " ); 
+        return false;
+    }
+    else {
+        struct timeval startTime,
+                       currentTime,
+                       duration;
+
+        gettimeofday ( & startTime, 0 );
+
+        while ( 1 ) {
+          sleep ( 5 );
+          int status;
+          int returned_pid = waitpid ( pid, &status, WNOHANG );
+          if ( returned_pid == pid ) {
+              int exit_status = WEXITSTATUS(status);
+              if ( exit_status ) {
+                cerr << "Perftest failed. exit_status was: " << exit_status;
+                return false;
+              }
+              else {
+                return true; // perftest succeeded.
+              }
+          }
+          else  {  // perftest has not yet completed. 
+              gettimeofday ( & currentTime, 0 );
+              timersub ( & currentTime, & startTime, & duration );
+              if ( duration.tv_sec > 60 ) {
+                kill ( pid, 9 );
+                cerr << "Perftest pid " << pid << " hanging: killed.\n";
+                return false;
+              }
+          }
+        }
+                       
+    }
+}
+
+
+
+bool
+allBrokersAreAlive ( brokerVector & brokers ) {
+    for ( unsigned int i = 0; i < brokers.size(); ++ i ) {
+        pid_t pid = brokers[i]->getPID();
+        int status;
+        int value;
+        if ( (value = waitpid ( pid, &status, WNOHANG ) ) ) {
+           return false; 
+        }
+    }
+
+    return true;
+}
+
+
+
+
+void
+killAllBrokers ( brokerVector & brokers ) {
+    for ( unsigned int i = 0; i < brokers.size(); ++ i )
+        brokers[i]->kill ( 9 );
+}
+
+
+
+
+}} // namespace qpid::tests
+
+using namespace qpid::tests;
+
+
+
+int
+main ( int argc, char ** argv )
+{
+    int n_iterations = argc > 0 ? atoi(argv[1]) : 1;
+    int n_brokers = 3;
+    brokerVector brokers;
+
+    for ( int i = 0; i < n_brokers; ++ i ) {
+        startBroker ( brokers, i );
+    }
+
+    sleep ( 3 );
+
+    /* Run all perftest iterations, and only then check for brokers 
+     * still being up.  If you just want a quick check for the failure 
+     * mode in which a single iteration would kill all brokers except 
+     * the client-connected one, just run it with the iterations arg
+     * set to 1.
+    */
+    for ( int iteration = 0; iteration < n_iterations; ++ iteration ) {
+        if ( ! runPerftest ( ) ) {
+            cerr << "Perftest " << iteration << " failed.\n";
+            return 1;
+        }
+        if ( ! ( iteration % 10 ) ) {
+            cerr << "perftest " << iteration << " complete. -------------- \n";
+        }
+    }
+    cerr << "\nperftest " << n_iterations << " iterations complete. -------------- \n\n";
+
+    if ( ! allBrokersAreAlive ( brokers ) ) {
+        cerr << "not all brokers are alive.\n";
+        return 2;
+    }
+
+    killAllBrokers ( brokers );
+    return 0;
+}
+
+
+
diff --git a/cpp/src/tests/cluster_test.cpp b/cpp/src/tests/cluster_test.cpp
index d07c0ecdb5..8c18e578df 100644
--- a/cpp/src/tests/cluster_test.cpp
+++ b/cpp/src/tests/cluster_test.cpp
@@ -54,6 +54,7 @@
 #include <algorithm>
 #include <iterator>
 
+
 using namespace std;
 using namespace qpid;
 using namespace qpid::cluster;