#!/usr/bin/env python
# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2008 - 2009, SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: component_container.py 9868 2009-11-21 17:50:45Z dhiraj $
import getopt
import time
import random
import urlparse
from paste import httpserver
from snaplogic import rp,snapi
from snaplogic import cc
from snaplogic.common.snap_exceptions import *
from snaplogic.common import snapstream,runtime_table
from snaplogic.common import snap_http_lib,snap_crypt
from snaplogic.common import version_info
from snaplogic.common.snap_http_lib import concat_paths,add_params_to_uri,setUsernamePassword,sendreq
from snaplogic.common import uri_prefix
from snaplogic.common import snap_log,snap_control,snap_stats
from snaplogic.common.config import snap_config,credentials_store
from snaplogic import snapi_base
from snaplogic.snapi_base import keys
from snaplogic.snapi_base.exceptions import SnapiException
from snaplogic.server.http_request import HttpRequest
from snaplogic.server import RhResponse
from snaplogic.cc import component_runtime
from snaplogic.cc import cc_info
from snaplogic.cc import registration
def get_logs(http_req):
"""
Return server log info.
@param http_req: HTTP request object.
@type http_req: L{HttpRequest}
@return: RhResponse object with data and code
to be written to client.
@rtype: L{RhResponse}
"""
resp = snap_log.get_logs(http_req, cc.my_process_uri, cc.logger, cc.rlog, cc.elog)
if resp:
return (resp.http_code, resp.data)
else:
return None
def get_stats(http_req):
"""
Return server stats information.
@param http_req: HTTP request object.
@type http_req: L{HttpRequest}
@return: RhResponse object with data and code
to be written to client.
@rtype: L{RhResponse}
"""
d = cc.stats.get_all_groups()
return (200, d)
def application(environ, start_response):
"""
Handle a new HTTP request.
This is the core function of our WSGI application.
@param environ: The environment object of the WSGI request.
It contains all HTTP headers.
@type environ: dictionary
@param start_response: A callable provided by WSGI, which can be used
to write the response.
@type start_response: callable
"""
cc.stats_group.get_stat("number_of_requests").inc()
http_req = HttpRequest(environ, start_response)
try:
ret = request_handler(http_req)
except SnapHttpErrorCode, e:
cc.elog(e, "Request %s %s failed" % (http_req.method, http_req.path))
ret = (e.http_code, e.err_message if e.err_message is not None else "")
cc.stats_group.get_stat("number_of_failed_requests").inc()
except Exception, e:
cc.elog(e, "Request %s %s failed" % (http_req.method, http_req.path))
ret = (http_req.INTERNAL_SERVER_ERROR, None)
cc.stats_group.get_stat("number_of_failed_requests").inc()
if ret:
(return_code, return_data) = ret
http_req.send_response_headers(return_code, None, True)
if return_data:
http_req.output.write(return_data)
http_req.output.end()
cc.rlog(http_req, ret[0])
else:
cc.rlog(http_req, http_req.OK)
return []
def request_handler(http_req):
"""
Implement the request handler got the CC.
This code implements the task of routing HTTP requests, based on their URI and HTTP
method, to the appropriate function that can process the request. This handler is
useful in clearly demonstrating how the URI space is carved up for the CC and in
showing which HTTP methods are allowed for those URIs.
@param http_req: The request object
@type http_req: L{HttpRequest}
@return: (http error code, data to be returned)
@rtype: 2-tuple
"""
ret = None
if http_req.method == "GET":
cc.stats_group.get_stat("number_of_get_requests").inc()
elif http_req.method == "POST":
cc.stats_group.get_stat("number_of_post_requests").inc()
elif http_req.method == "PUT":
cc.stats_group.get_stat("number_of_put_requests").inc()
elif http_req.method == "DELETE":
cc.stats_group.get_stat("number_of_delete_requests").inc()
if http_req.path.startswith(uri_prefix.RUNTIME):
# Resource runtime related requests.
if http_req.path == uri_prefix.RUNTIME_STATUS and http_req.method == "GET":
# We need to make an auth check to make sure that the token matches.
if cc.cc_token is not None and ("CC_TOKEN" not in http_req.snapi_headers or
http_req.snapi_headers["CC_TOKEN"] != cc.cc_token):
cc.stats_group.get_stat("number_of_unauthorized_requests").inc()
return (http_req.UNAUTHORIZED, "The token provided is invalid")
ret = cc_info.list_resource_runtimes(http_req)
elif http_req.method == "GET":
ret = component_runtime.process_runtime_get_request(http_req)
elif http_req.method == "PUT":
ret = component_runtime.process_runtime_put_request(http_req)
elif http_req.method == "POST":
ret = component_runtime.process_runtime_post_request(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Received request: %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
else:
# We need to make an auth check to make sure that the token matches.
if cc.cc_token is not None and ("CC_TOKEN" not in http_req.snapi_headers or
http_req.snapi_headers["CC_TOKEN"] != cc.cc_token):
return (http_req.UNAUTHORIZED, "The token provided is invalid")
if http_req.path.startswith(uri_prefix.COMPONENT_RESOURCE_TEMPLATE):
# Its a fetch resource template request
if http_req.method == "POST":
ret = cc_info.create_resource_template(http_req)
elif http_req.method == "GET":
ret = cc_info.get_resource_template(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Create resource template URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
elif http_req.path.startswith(uri_prefix.COMPONENT_UPGRADE_RESOURCE):
# Upgrade the resource
if http_req.method == "POST":
ret = cc_info.upgrade_resource(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Upgrade resource URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
elif http_req.path.startswith(uri_prefix.COMPONENT_SUGGEST_RESOURCE_VALUES):
# Its a suggest resource values request
if http_req.method == "POST":
ret = cc_info.suggest_resource_values(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Suggest resource value URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
elif http_req.path.startswith(uri_prefix.COMPONENT_VALIDATE_RESOURCE):
# Validate a resdef with the component.
if http_req.method == "POST":
cc.stats_group.get_stat("number_of_resdef_validation_requests").inc()
ret = cc_info.validate_resdef(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Validate resource URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
elif http_req.path.startswith(uri_prefix.COMPONENT_LIST):
# GET component list.
if http_req.method == "GET":
ret = cc_info.list_components(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "List components URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
elif http_req.path.startswith(uri_prefix.LOGS):
# GET the logs from the server.
if http_req.method == "GET":
ret = get_logs(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Getting logs URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
elif http_req.path.startswith(uri_prefix.STATS):
# GET the stats from the server.
if http_req.method == "GET":
ret = get_stats(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "Getting stats URI received %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.METHOD_NOT_ALLOWED, "Method %s is not supported by this URI" % http_req.method)
else:
if http_req.method == "POST":
# Must be prepare request for a component
ret = component_runtime.process_prepare_request(http_req)
else:
cc.log(snap_log.LEVEL_ERR, "CC received request: %s with unsupported method %s" %
(http_req.path, http_req.method))
ret = (http_req.NOT_FOUND, "Unknown request")
if ret:
(code, txt) = ret
if code == http_req.METHOD_NOT_ALLOWED:
cc.stats_group.get_stat("number_of_invalid_method_requests").inc()
return ret
def usage():
print """component_container.py --name=<name of component container> --server=<server URI>
--credentials <credentials file name> --disable-token
--component_conf_dir=<path> --set <config_param=value>
The token parameter is only needed for test purposes
"""
def main():
# Process command line.
cc_name = server_uri = cred_store = cred = None
cc.cc_token = snap_crypt.generate_random_string()
# CC config parameters may be specified via command line
# in which case they override parameters coming from snapserver.conf CC section.
override_config = {}
try:
opts, xargs = getopt.getopt(sys.argv[1:], "n:s:c:h",
["name=", "server=", "credentials=", "disable-token", "component_conf_dir=", "set=", "help"])
except getopt.GetoptError, e:
usage()
sys.exit("Option(s) specified are not valid %s" % str(e))
for o, a in opts:
if o in ("-n", "--name"):
cc_name = a
elif o in ("-s", "--server"):
server_uri = a
elif o in ("-c", "--credentials"):
cred_store = a
elif o in ("--disable-token"):
cc.cc_token = None
# Alternative component configuaration dir. Overrides default setting
elif o in ("--component_conf_dir",):
override_config['component_conf_dir'] = a
elif o == '--set':
# Config parameters may be overridden using command line --set e.g.:
# --set cc_http_proxy=myproxy
try:
(param, value) = a.split('=')
except Exception, e:
usage()
sys.exit(e.message)
override_config[param] = value
elif o in ("-h", "--help"):
usage()
sys.exit()
if cc_name is None:
usage()
sys.exit('Error: A Component Container name is required.')
else:
cc.cc_name = cc_name
if cred_store is not None:
credentials_store.initialize(cred_store)
cred = credentials_store.get_credentials_by_uri("*")
if cred is None:
sys.exit('Error: Username and password not found in credentials file.')
else:
# Store the username, this can be used in certain situations. Don't store the password
# for security reasons.
cc.cc_username = cred[0]
if server_uri is None:
usage()
sys.exit('Error: The server URI is required.')
# We get our config from the main server in the form of a simple dictionary.
config = fetch_config(server_uri, cc_name, cred)
if config == None:
sys.exit('Error: Failed to get config.')
# If any configuration parameters were specified via the command line override them
for param in override_config:
config[param] = override_config[param]
runtime_table.initialize(config['runtime_entry_timeout'])
# Create the stats for this process
cc.stats_group = snap_stats.SnapStatsGroup("Request Handler")
cc.stats_group.add_field_stat("alive_since", "number", time.time())
cc.stats_group.add_counter_stat("number_of_requests")
cc.stats_group.add_counter_stat("number_of_failed_requests")
cc.stats_group.add_counter_stat("number_of_unauthorized_requests")
cc.stats_group.add_counter_stat("number_of_get_requests")
cc.stats_group.add_counter_stat("number_of_post_requests")
cc.stats_group.add_counter_stat("number_of_put_requests")
cc.stats_group.add_counter_stat("number_of_delete_requests")
cc.stats_group.add_counter_stat("number_of_invalid_method_requests")
cc.stats_group.add_counter_stat("number_of_resdef_validation_requests")
cc.stats_group.add_counter_stat("number_of_spawned_component_threads")
try:
cc.stats = snap_stats.SnapStats()
except:
cc.stats = snap_stats.get_instance()
cc.stats.add_stats_group(cc.stats_group)
# The URI of this process is stored in the config object. However, because we need
# our own URI quite often, we also store it directly in the CC object, so that
# we can get to it quickly and without having to do a lookup in the config object.
cc.my_listening_uri = None
if "cc_proxy_uri" in config and config["cc_proxy_uri"]:
cc.my_process_uri = config["cc_proxy_uri"]
else:
if not "cc_hostname" in config or not config["cc_hostname"]:
raise SnapException("Config entry 'cc_hostname' must be set if 'cc_proxy_uri' was not defined.")
cc.my_listening_uri = "http://%s:%s" % (config["cc_hostname"], config["cc_port"])
cc.my_process_uri = cc.my_listening_uri
# Now we can transcribe whatever we got into a nice SnapConfig object. Note
# the slightly odd way of doing this: We specify a ready-made dictionary of
# sections (we define the 'cc' section here). At the same time, we pass a
# second dictionary. This dictionary (the third parameter) is a bit special,
# because it is for the 'common' section in the config. It has a fixed name
# and meaning, which is why it is specified in that separate manner.
try:
cc.config = snap_config.SnapConfig(None, { "cc" : config }, snap_config.CC_DEFAULT,
{ "my_process_uri" : cc.my_process_uri,
"main_process_uri" : server_uri,
"i_am_main_process": False })
except Exception, e:
raise e
sys.exit('Error: Failed to establish config object...')
cc_conf = cc.config.get_section('cc')
if not cc.my_listening_uri:
cc.my_listening_uri = "http://%s:%s" % (cc_conf["cc_address"], cc_conf["cc_port"])
# Initialize log
cr = 'no'
try:
cr = cc_conf["console_output"]
to_console = cr == 'yes'
except:
to_console = False
if cr not in [ 'yes', 'no' ]:
raise SnapException("Config entry 'console_output' must be either 'yes' or 'no', found '%s' instead." % cr)
cc.init_loggers(cc_name, cc_conf['log_dir'], cc_conf['log_level'], to_console, cc_conf['disable_logging'],
cc_conf["max_log_size"], cc_conf["log_backup_count"])
print version_info.server_banner
print version_info.server_copyright
print "Starting Component Container '%s'..." % cc_name
parsed_uri = urlparse.urlparse(cc.my_listening_uri)
if snap_http_lib.is_localhost(parsed_uri[1].split(":")[0]):
cc.log(snap_log.LEVEL_WARN,
"CC configuration parameter 'cc_uri' is set to '%s'. Network capabilities will be restricted."
% cc.my_listening_uri)
if snap_http_lib.is_localhost(cc_conf["cc_address"]):
cc.log(snap_log.LEVEL_WARN,
"CC configuration parameter 'cc_address' is set to '%s'. Network capabilities will be restricted."
% cc_conf["cc_address"])
# Initialize SnapStream from config
snapstream.init(cc.config)
# Initialize component runtime from config
component_runtime.init(cc_conf)
snapi._initialize(server_uri)
cc.server_uri = server_uri
# Load configured components
registration.discover_configured_components()
cc.log(snap_log.LEVEL_INFO, "Component Container %s listening on %s:%s" % (cc_name, cc_conf["cc_address"], config["cc_port"]))
# Tell the HttpRequest class that it should never allow human readable requests.
# This is useful so that we don't get confused if someone accidentally points a
# browser at us: We don't have all the information necessary to display human
# readable output. Only the main server has that.
HttpRequest.allow_human_readable = False
# Enter the listening loop
print "Component Container '%s' started." % cc_name
httpserver.serve(application, host=cc_conf["cc_address"], port=int(config["cc_port"]), use_threadpool=False)
def fetch_config(server_uri, name, credential):
"""
Retrieve config file from main process.
The CC does not have a config file of its own. It depends upon the main server process
to have access to this information. This approach, ensures that there is only one process
responsible for reading the file. This function enters a loop, attempting to GET the file
from the server and it retries until it gets the config file.
@param server_uri: URI of the main server process.
@type server_uri: str
@param name: Name of the component container.
@type name: str
@return: The config sent by server in dictionary format.
@rtype: dict
"""
server_uri = concat_paths(server_uri, uri_prefix.CC_REGISTER)
if cc.cc_token is None:
# We have disabled token authentication on CC side. Use a place holder value.
tk = cc.PLACE_HOLDER_CC_TOKEN
else:
tk = cc.cc_token
request_body = {keys.CC_NAME : name,
keys.CC_TOKEN : tk}
while True:
try:
resp_dict = snapi_base.send_req("POST", server_uri, request_body, None, credential)
except SnapiException, e:
raise SnapIOError("Config fetch from main server failed. %s" % e)
except Exception, e:
print e
time.sleep(5)
continue
return resp_dict
if __name__ == "__main__":
main()
|