Code cleanup, updated datastore interface, removed unused files

AppController/lib/haproxy.rb

@@ -171,22 +171,22 @@ def self.write_app_config(app_name, app_number, num_of_servers, ip)
   # Updates the HAProxy config file for this App Engine application to
   # point to all the ports currently the application. In contrast with
   # write_app_config, these ports can be non-contiguous.
-  # TODO(cgb): Lots of copypasta here with write_app_config - eliminate it.
-  def self.update_app_config(app_name, app_number, ports, public_ip)
+  # TODO(cgb): Lots of copy/paste here with write_app_config - eliminate it.
+  def self.update_app_config(app_name, app_number, ports, private_ip)
     # Add a prefix to the app name to avoid collisions with non-GAE apps
     full_app_name = "gae_#{app_name}"
     index = 0
     servers = []
 
     ports.each { |port|
-      server = HAProxy.server_config(full_app_name, index, public_ip, port)
+      server = HAProxy.server_config(full_app_name, index, private_ip, port)
       index += 1
       servers << server
     }
 
     listen_port = HAProxy.app_listen_port(app_number)
     config = "# Create a load balancer for the app #{app_name} \n"
-    config << "listen #{full_app_name} #{public_ip}:#{listen_port} \n"
+    config << "listen #{full_app_name} #{private_ip}:#{listen_port} \n"
     config << servers.join("\n")
 
     config_path = File.join(SITES_ENABLED_PATH, 

AppController/lib/pbserver.rb

@@ -47,6 +47,9 @@ module PbServer
   # within AppScale.
   DBS_WITH_NATIVE_PBSERVER = ["mysql"]
 
+  # A list of databases that use the version 2 of database query support.
+  # The second version has secondary index support.
+  DBS_WITH_V2_PBSERVER = ['cassandra', 'hypertable', "hbase"]
 
   # The name that nginx should use as the identifier for the PBServer when it
   # we write its configuration files.
@@ -64,14 +67,26 @@ def self.start(master_ip, db_local_ip, my_ip, table, zklocations)
       "MASTER_IP" => master_ip, 
       "LOCAL_DB_IP" => db_local_ip 
     }
-
-    ports.each { |port|
-      start_cmd = "/usr/bin/python2.6 #{pbserver} -p #{port} " +
+
+    if DBS_WITH_V2_PBSERVER.include?(table)
+      ports.each { |port|
+        start_cmd = "/usr/bin/python2.6 #{pbserver} -p #{port} " +
+            "--no_encryption --type #{table} -z \'#{zklocations}\' "
+        # stop command doesn't work, relies on terminate.rb
+        stop_cmd = "pkill -9 datastore_server"
+        GodInterface.start(:pbserver, start_cmd, stop_cmd, port, env_vars)
+      }
+    else
+      ports.each { |port|
+        start_cmd = "/usr/bin/python2.6 #{pbserver} -p #{port} " +
+            "--no_encryption --type #{table} -z \'#{zklocations}\' "
           "--no_encryption --type #{table} -z \'#{zklocations}\' " +
           "-s #{HelperFunctions.get_secret()} -a #{my_ip} --key"
-      stop_cmd = "pkill -9 appscale_server"
-      GodInterface.start(:pbserver, start_cmd, stop_cmd, port, env_vars)
-    }
+        # stop command doesn work, relies on terminate.rb
+        stop_cmd = "pkill -9 appscale_server"
+        GodInterface.start(:pbserver, start_cmd, stop_cmd, port, env_vars)
+      }
+    end
   end
 
 
@@ -117,6 +132,8 @@ def self.is_running(my_ip)
   def self.get_executable_name(table)
     if DBS_WITH_NATIVE_PBSERVER.include?(table)
       return "#{APPSCALE_HOME}/AppDB/appscale_server_#{table}.py"
+    elsif DBS_WITH_V2_PBSERVER.include?(table)
+      return "#{APPSCALE_HOME}/AppDB/datastore_server.py"
     else
       return "#{APPSCALE_HOME}/AppDB/appscale_server.py"
     end

AppController/terminate.rb

@@ -113,7 +113,7 @@
 
 ["memcached",
  "nginx", "haproxy", "collectd", "collectdmon",
- "soap_server", "appscale_server",
+ "soap_server", "appscale_server", "datastore_server",
  "AppLoadBalancer", "AppMonitoring",
  # AppServer
  "dev_appserver", "DevAppServerMain",

AppDB/appscale_datastore.py

@@ -5,24 +5,21 @@
 # Author: Gaurav Kumar Mehta
 # Author: NOMURA Yoshihide
 # See LICENSE file
-import threading
+import imp
+import os
 import sys
-import string, cgi
+import string
 import socket
-import os
+import threading
 import types
-import imp
+
 import appscale_logger
 from dbconstants import *
 
-#from helper_functions import ThreadedLogger
-#LOG_DIR = "%s/AppDB/logs" % APPSCALE_HOME
-#LOG_FILENAME = LOG_DIR + "/appscale_datastore.log"
-#app_datastore_logger = ThreadedLogger(LOG_FILENAME)
-#app_datastore_logger.turnLoggingOn()
 app_datastore_logger = appscale_logger.getLogger("appscale_datastore")
 
 DB_ERROR = "DB_ERROR:"
+
 ERROR_CODES = [DB_ERROR]
 
 DATASTORE_DIR= "%s/AppDB" % APPSCALE_HOME
@@ -38,8 +35,10 @@ def getDatastore(cls, d_type):
       d_mod = imp.load_source(d_name, mod_path)
       datastore = d_mod.DatastoreProxy(app_datastore_logger)
     else:
-      app_datastore_logger.error("Fail to use datastore: %s. Please check the datastore type." % d_type)
-      raise Exception("Fail to use datastore: %s" % d_type)
+      app_datastore_logger.error("Fail to use datastore: %s. Please " + \
+                                 "check the datastore type." % d_type)
+      raise Exception("Datastore was not found in %d directory. " + \
+                      "Fail to use datastore: %s" %(DATASTORE_DIR, d_type))
     return datastore
 
   @classmethod

AppDB/appscale_datastore_batch.py

@@ -0,0 +1,54 @@
+#!/usr/bin/python
+# Programmer: Navraj Chohan <nlake44@gmail.com>
+# See LICENSE file
+
+import imp
+import os
+import socket
+import string
+import sys
+import threading
+import types
+
+import appscale_logger
+import dbconstants
+
+app_datastore_logger = appscale_logger.getLogger("appscale_datastore_batch")
+
+DATASTORE_DIR= "%s/AppDB" % dbconstants.APPSCALE_HOME
+
+class DatastoreFactory:
+
+  @classmethod
+  def getDatastore(cls, d_type):
+    """ Returns a reference for the datastore. Validates where 
+        the <datastore>_interface.py is and adds that path to 
+        the system path.
+
+    Args: 
+      d_type: The name of the datastore (ex: cassandra)
+    """
+
+    datastore = None
+    mod_path = DATASTORE_DIR + "/" + d_type + "/" + d_type + "_interface.py"
+
+    if os.path.exists(mod_path):
+      sys.path.append(DATASTORE_DIR + "/" + d_type)
+      d_mod = imp.load_source(d_type+"_interface.py", mod_path)
+      datastore = d_mod.DatastoreProxy(app_datastore_logger)
+    else:
+      app_datastore_logger.error("Fail to use datastore: %s. Please \
+                                  check the datastore type." % d_type)
+      raise Exception("Fail to use datastore: %s" % d_type)
+
+    return datastore
+
+  @classmethod
+  def valid_datastores(cls):
+    """ Returns a list of directories where the datastore code is
+
+    Returns: Directory list 
+    """
+
+    dblist = os.listdir(DATASTORE_DIR)
+    return dblist

AppDB/appscale_server.py

@@ -956,8 +956,6 @@ def put_request(self, app_id, http_request_data):
 
     start_time = time.time() 
     putreq_pb = datastore_pb.PutRequest(http_request_data)
-    #print "RECEIVED PUT_REQUEST %s" % putreq_pb
-    #logger.debug("RECEIVED PUT_REQUEST %s" % putreq_pb)
     putresp_pb = datastore_pb.PutResponse( )
     txn = None
     root_key = None
@@ -1039,7 +1037,6 @@ def put_request(self, app_id, http_request_data):
         # then the global counter is used
         # gen unique id only wants to know if a child exist
         uid = generate_unique_id(app_id, root_key, child_key)
-        #print "UID:" + str(uid)
         if uid <= 0: 
           return(putresp_pb.Encode(), 
                  datastore_pb.Error.INTERNAL_ERROR, 
@@ -1078,7 +1075,6 @@ def put_request(self, app_id, http_request_data):
                   'No group entity or root key.')
         try:          
           locktime = time.time()
-          print root_key
           gotLock = zoo_keeper.acquireLock( app_id, txn.handle(), root_key)
           if PROFILE: appscale_log.write("ACQUIRELOCK %d %f\n"%(txn.handle(), time.time() - locktime))
         except zk.ZKTransactionException, zkex:
@@ -1092,7 +1088,6 @@ def put_request(self, app_id, http_request_data):
       #######################################
       # insert key 
       table_name = getTableName(app_id, kind, namespace)
-      #print "Put Using table name:",table_name
       # Notify Users/Apps table if a new class is being added 
       if table_name not in tableHashTable:
         # This is the first time this pbserver has seen this table
@@ -1171,7 +1166,6 @@ def put_request(self, app_id, http_request_data):
                          JOURNAL_SCHEMA,
                          [e.Encode()])
       entPut = putThread()
-      #print "Row key:" + str(row_key)
       handle = 0
       if is_trans_on: 
         handle = txn.handle() 
@@ -1220,7 +1214,6 @@ def get_request(self, app_id, http_request_data):
     global app_datastore
     getreq_pb = datastore_pb.GetRequest(http_request_data)
     #logger.debug("GET_REQUEST: %s" % getreq_pb)
-    #print "GET_REQUEST: %s" % getreq_pb
     getresp_pb = datastore_pb.GetResponse()
 
     is_trans_on = True
@@ -1262,9 +1255,6 @@ def get_request(self, app_id, http_request_data):
         zoo_keeper = zoo_keeper_stub
       table_name = getTableName(app_id, kind, namespace)
       row_key = getRowKey(app_id,key.path().element_list())
-      #print "get row key:" + str(row_key)
-      #print "table_name:" + str(table_name)
-      #print "schema:" + str(ENTITY_TABLE_SCHEMA)
       r = app_datastore.get_entity( table_name, row_key, ENTITY_TABLE_SCHEMA )
       err = r[0]
       if err not in ERROR_CODES or len(r) != 3: 
@@ -1482,35 +1472,21 @@ def optimized_put_request(self, app_id, http_request_data):
 
   def void_proto(self, app_id, http_request_data):
     resp_pb = api_base_pb.VoidProto() 
-    print "Got void"
-    #logger.debug("VOID_RESPONSE: %s to void" % resp_pb)
     return (resp_pb.Encode(), 0, "" )
 
   def str_proto(self, app_id, http_request_data):
     str_pb = api_base_pb.StringProto( http_request_data )
     composite_pb = datastore_pb.CompositeIndices()
-    print "Got a string proto"
-    print str_pb
-    #logger.debug("String proto received: %s"%str_pb)
-    #logger.debug("CompositeIndex response to string: %s" % composite_pb)
     return (composite_pb.Encode(), 0, "" )    
 
   def int64_proto(self, app_id, http_request_data):
     int64_pb = api_base_pb.Integer64Proto( http_request_data ) 
     resp_pb = api_base_pb.VoidProto()
-    print "Got a int 64"
-    print int64_pb
-    #logger.debug("Int64 proto received: %s"%int64_pb)
-    #logger.debug("VOID_RESPONSE to int64: %s" % resp_pb)
     return (resp_pb.Encode(), 0, "")
 
   def compositeindex_proto(self, app_id, http_request_data):
     compindex_pb = entity_pb.CompositeIndex( http_request_data)
     resp_pb = api_base_pb.VoidProto()
-    print "Got Composite Index"
-    #print compindex_pb
-    #logger.debug("CompositeIndex proto recieved: %s"%str(compindex_pb))
-    #logger.debug("VOID_RESPONSE to composite index: %s" % resp_pb)
     return (resp_pb.Encode(), 0, "")
 
 # Returns 0 on success, 1 on failure
@@ -1526,28 +1502,22 @@ def create_index_tables(self, app_id):
       return 1
     """
     table_name = "__" + app_id + "__" + "single_prop_asc"
-    print "Building table: " + table_name
     columns = ["reference"]
     returned = app_datastore.create_table( table_name, columns )
     err,res = returned
     if err not in ERROR_CODES:
-      #logger.debug("%s" % err)
       return 1
 
     table_name = "__" + app_id + "__" + "single_prop_desc"
-    print "Building table: " + table_name
     returned = app_datastore.create_table( table_name, columns )
     err,res = returned
     if err not in ERROR_CODES:
-      #logger.debug("%s" % err)
       return 1
 
     table_name = "__" + app_id + "__" + "composite"
-    print "Building table: " + table_name
     returned = app_datastore.create_table( table_name, columns )
     err,res = returned
     if err not in ERROR_CODES:
-      #logger.debug("%s" % err)
       return 1 
 
     return 0