# $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: snap_control.py 10183 2009-12-16 23:58:55Z grisha $
"""
SnapLogic Pipeline management related code
"""
import random
import copy
from distutils.version import StrictVersion
from snaplogic.common import snap_http_lib
from snaplogic.common.snap_exceptions import *
import snaplogic.common.resource_runtime as resource_runtime
from snaplogic.common.runtime_status import RuntimeStatus
from snaplogic.snapi_base import keys
from snaplogic.snapi_base import send_list_req
from snaplogic import rp
# We currently use only JSON for SnapControl messages
plugin = rp.get_rp('application/json')
SNAP_CONTROL_MAJOR = 1
SNAP_CONTROL_MINOR = 1
SNAP_CONTROL_VERSION = "%s.%s" % (SNAP_CONTROL_MAJOR, SNAP_CONTROL_MINOR)
def compare_versions(vers):
"""
Compare the remote version with the version supported.
@param vers: Versions to as a string in the format N.M, where N and M are positive integers.
@type vers: str
@return: A 2-tuple. The first entry will be True, if the two versions are compatible in major numbers. The
second entry will be 1, 0, -1, depending on whether the remote version's minor number is greater, equal
or lesser than the version supported by the server.
@rtype: 2-tuple.
"""
r = vers.split(".")
if len(r) != 2:
raise SnapValueError("Version number \"%s\" is invalid. Should be N.M, where N and M are positive integers"
% vers)
try:
maj = int(r[0])
except:
raise SnapValueError("Version major number in \"%s\" is invalid. "
"Should be N.M, where N and M are positive integers"
% vers)
try:
min = int(r[1])
except:
raise SnapValueError("Version minor number \"%s\" is invalid. "
"Should be N.M, where N and M are positive integers"
% vers)
if maj != SNAP_CONTROL_MAJOR:
return (False, None)
elif min > SNAP_CONTROL_MINOR:
return (True, 1)
elif min < SNAP_CONTROL_MINOR:
return (True, -1)
else:
return (True, 0)
def init_message():
"""
Return a basic template for SnapControl message.
@return: Message dictionary.
@rtype: dict
"""
mesg = {}
mesg["snap_control_version"] = SNAP_CONTROL_VERSION
return mesg
def get_non_null_value(dict, key, *err_msg):
"""
Retrieve value from dictionary by key. The value cannot be None.
@param dict: The dictionary being read.
@type dict: dict
@param key: The key being looked up.
@type key: str
@param *err_msg: Error message to be used in exception, if value is not found.
@type *err_msg: List of strings.
"""
if key not in dict or dict[key] is None:
e = " ".join(err_msg)
raise SnapValueError("%s: \"%s\" not found." % (e, key))
return dict[key]
def get_value(dict, key, *err_msg):
"""
Retrieve value from dictionary by key.
@param dict: The dictionary being read.
@type dict: dict
@param key: The key being looked up.
@type key: str
@param *err_msg: Error message to be used in exception, if value is not found.
@type *err_msg: List of strings.
"""
if key not in dict:
e = " ".join(err_msg)
raise SnapValueError("%s: \"%s\" not found." % (e, key))
return dict[key]
def create_prepare_request(res_rt):
"""
Create PREPARE request message.
This method is called by the PM to prepare a resource for execution. The method uses the information
about the resource in the resource runtime object to create the PREPARE request.
Sample format of the PREPARE request:
{
"snap_control_version": "1.0",
"pipeline_runtime_identifier" : "1233.4556"
"resource_name" : "reader"
"context_name" : "StockPipe.SortedSource.ReaderPipe" # This is an optional entry in the dictionary.
"parameters" : {
"name1" : "value1",
"name2" : "value2",
"name3" : "value3"
},
"resource_ref" : { "DBConnection1" : { "uri" : "uri of DB connection resource #1",
"parameters" : { "param1" : "value1",
"param2" : "value2"},
"DBConnection2" : { "uri" : "uri of DB connection resource #2",
"parameters" : { "param3" : "value3",
"param4" : "value4"}
},
"input_views" : {
"inp1": {
"method": "GET",
"uri": "http://moxie:8088/__snap__/runtime/view/123456768",
"field_links": [
["src_field1", ["dest_field1", "dest_field2"]],
["src_field2", []],
["src_field3", ["dest_field3"]]
],
},
"inp2": {
"method" : "POST",
# "exact_match" means that the upstream component will send data in the exact format
# defined by the input view. In effect, no real mapping needs to be done.
"field_links" : "exact_match",
},
"inp3" : {
"method" : "GET",
}
},
"output_views" : {
"out1": {
"method" : "GET",
},
"out2": {
"method" : "GET"
}
}
}
@param res_rt: The resource runtime object being used to create the prepare request.
@type res_rt: L{resource_runtime.ResourceRuntime}
@return: The request message.
@rtype: dict
"""
mesg = init_message()
mesg["pipeline_runtime_identifier"] = res_rt.pipeline_rid
mesg["resource_name"] = res_rt.resource_name
if res_rt.context_name is not None:
mesg["context_name"] = res_rt.context_name
mesg["parameters"] = res_rt.parameters
if hasattr(res_rt, "resource_ref") and res_rt.resource_ref:
mesg[keys.RESOURCE_REF] = {}
for (ref_name, ref) in res_rt.resource_ref.items():
mesg[keys.RESOURCE_REF][ref_name] = copy.deepcopy(ref)
mesg["input_views"] = {}
for inp in res_rt.input_views.values():
i = mesg["input_views"][inp.name] = {}
i["method"] = inp.http_method
if i["method"].upper() == "GET":
# For GET method of feeding input view, a URI must also be specified.
i["uri"] = inp.uri
elif i["method"].upper() != "POST":
raise SnapValueError("Input view \"%s\" has unexpected http method \"%s\"" % (inp.name, inp.http_method))
i["field_links"] = inp.field_links
mesg["output_views"] = {}
for out in res_rt.output_views.values():
o = mesg["output_views"][out.name] = {}
o["method"] = "GET"
if out.pass_through_fields is not None:
# We have specifications for the pass through fields.
o["pass_through_fields"] = out.pass_through_fields
return mesg
def parse_prepare_request(mesg, res_rt):
"""
Parse the prepare request and populate resource runtime object based on the request.
@param mesg: The prepare request SnapControl message.
@type mesg: dict
@param: The runtime resource object being populated.
@type: L{resource_runtime.ResourceRuntime}
"""
# The runtime identifier of a resource is a dot separated alpha numeric value. This value is derived by
# taking the runtime identifier of the pipeline (if any) that executed a resource and appending a period,
# followed by some unique alpha numeric value.
val = get_value(mesg, "pipeline_runtime_identifier", "Prepare request invalid")
if val:
rid = mesg["pipeline_runtime_identifier"] + "." + str(random.randint(1,999999999))
pipeline_rid = mesg["pipeline_runtime_identifier"]
else:
# This is a very big number! The reason for this is that we really want those things
# to be unique. The client uses the pipeline ID as access token to the pipeline
# specific logs, for example. We don't want them to just be able to easily guess
# the IDs of other people's pipelines.
rid = str(random.randint(1,9999999999999999999999999999999999999999999999))
pipeline_rid = None
res_rt.resource_name = get_value(mesg, "resource_name", "Prepare request invalid")
res_rt.rid = rid
res_rt.pipeline_rid = pipeline_rid
if "parameters" in mesg:
res_rt.parameters = mesg["parameters"]
if "context_name" in mesg:
res_rt.context_name = mesg["context_name"]
# Populate the resource refs, if any.
res_rt.resource_ref = {}
if keys.RESOURCE_REF in mesg and mesg[keys.RESOURCE_REF]:
for (ref_name, ref) in mesg[keys.RESOURCE_REF].items():
# Assign the dict containing { keys.URI : http://...,
# keys.PARAMS : {...}}
res_rt.resource_ref[ref_name] = ref
if "input_views" in mesg:
for inp_name in mesg["input_views"]:
inp = res_rt.input_views[inp_name] = resource_runtime.RuntimeInputView(inp_name)
err_str = "Prepare request invalid. Input view %s" % inp_name
inp.http_method = get_non_null_value(mesg["input_views"][inp_name], "method",
"Invalid prepare request to ", res_rt.resource_uri,
". Input view ", inp_name).upper()
if inp.http_method == "GET":
inp.uri = get_non_null_value(mesg["input_views"][inp_name], "uri",
"Invalid prepare request to ", res_rt.resource_uri,
". Input view ", inp_name)
if "field_links" in mesg["input_views"][inp_name]:
inp.field_links = mesg["input_views"][inp_name]["field_links"]
if "output_views" in mesg:
for out_name in mesg["output_views"]:
out = res_rt.output_views[out_name] = resource_runtime.RuntimeOutputView(out_name)
out.http_method = get_non_null_value(mesg["output_views"][out_name], "method",
"Invalid prepare request to ", res_rt.resource_uri,
". Output view ", out_name).upper()
if "pass_through_fields" in mesg["output_views"][out_name]:
out.pass_through_fields = mesg["output_views"][out_name]["pass_through_fields"]
def create_prepare_response(res_rt):
"""
Create PREPARE response message.
This method is called by a CC or PM which has just received a PREPARE request and is trying
to send a response document in the same POST session.
Sample format of the PREPARE response:
{
"snap_control_version" : "1.0",
"runtime_identifier" : "1233.45556.78910"
"runtime_status_uri" : "http://moxie:8088/__snap__/runtime/status/entry/7mUy/resdef/csvread",
"runtime_control_uri" : "http://moxie:8088/__snap__/runtime/status/entry/control/Xqy5N/6e49/resdef/csvread",
"state_with_time_stamp" : ["Prepared", 1175287080],
"input_views": {
"inp2": {
"method": "POST",
"post_uri": "http://moxie:8088/__snap__/runtime/view/100032327",
"mode": "binary"
}
},
"output_views": {
"out1": {
"uri": "http://moxie:8088/__snap__/runtime/view/32842433",
"method": "GET",
"mode": "record"
},
"out2": {
"uri": "http://moxie:8088/__snap__/runtime/view/003243343232",
"method": "GET",
"mode": "binary"
}
}
}
@param res_rt: The resource runtime object being used to create the prepare response.
@type res_rt: L{resource_runtime.ResourceRuntime}
@return: The response message.
@rtype: dict
"""
mesg = init_message()
mesg["runtime_identifier"] = res_rt.rid
mesg[keys.RUNTIME_STATUS_URI] = res_rt.runtime_status_uri
mesg[keys.RUNTIME_CONTROL_URI] = res_rt.runtime_control_uri
mesg["state_with_time_stamp"] = res_rt.status.get_state_with_ts()
mesg["input_views"] = {}
for inp in res_rt.input_views.values():
if inp.http_method == "POST":
mesg["input_views"][inp.name] = {}
mesg["input_views"][inp.name]["method"] = inp.http_method
mesg["input_views"][inp.name]["uri"] = inp.uri
if inp.is_record:
mesg["input_views"][inp.name]["mode"] = "record"
else:
mesg["input_views"][inp.name]["mode"] = "binary"
mesg["output_views"] = {}
for out in res_rt.output_views.values():
mesg["output_views"][out.name] = {}
mesg["output_views"][out.name]["method"] = out.http_method
mesg["output_views"][out.name]["uri"] = out.uri
if out.is_record:
mesg["output_views"][out.name]["mode"] = "record"
else:
mesg["output_views"][out.name]["mode"] = "binary"
return mesg
def parse_prepare_response(mesg, res_rt):
"""
Parse prepare response from SnapControl message to resource runtime object.
@param mesg: SnapCotrol message.
@type mesg: dict
@param res_rt: Resource runtime object to be populated.
@type res_rt: L{resource_runtime.ResourceRuntime}
"""
res_rt.rid = mesg["runtime_identifier"]
# Prior to version 1.1, there is no difference between
# status and control URIs
resp_version = mesg["snap_control_version"]
if StrictVersion(resp_version) >= StrictVersion("1.1"):
res_rt.runtime_control_uri = mesg[keys.RUNTIME_CONTROL_URI]
res_rt.runtime_status_uri = mesg[keys.RUNTIME_STATUS_URI]
else:
res_rt.runtime_status_uri = mesg[keys.RUNTIME_STATUS_URI]
res_rt.runtime_control_uri = mesg[keys.RUNTIME_STATUS_URI]
state_ts = get_non_null_value(mesg, "state_with_time_stamp", "Status response invalid")
if state_ts[0] != RuntimeStatus.Prepared:
raise SnapValueError("Prepare response for resource %s (%s) had unexpected state value \"%s\"" %
(res_rt.resource_uri, res_rt.resource_name, state_ts[0]))
res_rt.status.set_state_with_ts(state_ts[0], state_ts[1])
if "input_views" in mesg:
for (inp_name, inp_dict) in mesg["input_views"].iteritems():
if inp_name not in res_rt.input_views:
raise SnapObjNotFoundError("Prepare response for resource %s (%s) had unknown input view \"%s\"" %
(res_rt.resource_uri, res_rt.resource_name, inp_name))
inp = res_rt.input_views[inp_name]
val = get_non_null_value(inp_dict, "method", "Invalid prepare response from ", res_rt.resource_uri,
". Input view ", inp_name).upper()
if inp.http_method != val:
raise SnapValueError("Prepare response for resource %s (%s) has method %s for input view \"%s\", "
"when http method expected was %s" %
(res_rt.resource_uri, res_rt.resource_name, val,
inp_name, inp.http_method))
if inp.http_method != "POST":
raise SnapValueError("Prepare response for resource %s (%s) provided a URI for input view \"%s\", "
"when http method was \"%s\"" %
(res_rt.resource_uri, res_rt.resource_name, inp_name, inp.http_method))
inp.uri = get_non_null_value(inp_dict, "uri", "Invalid prepare response from ", res_rt.resource_uri,
". Input view ", inp_name)
mode = get_non_null_value(inp_dict, "mode", "Invalid prepare response from ", res_rt.resource_uri,
". Input view ", inp_name).lower()
if mode == "record":
inp.is_record = True
elif mode == "binary":
inp.is_record = False
else:
raise SnapValueError("Prepare response for resource %s (%s) provided a mode %s for input view \"%s\""
% (res_rt.resource_uri, res_rt.resource_name, mode, inp_name))
if "output_views" in mesg:
for (out_name, out_dict) in mesg["output_views"].iteritems():
if out_name not in res_rt.output_views:
raise SnapObjNotFoundError("Prepare response for resource %s (%s) had unknown output view \"%s\"" %
(res_rt.resource_uri, res_rt.resource_name, out_name))
out = res_rt.output_views[out_name]
val = get_non_null_value(out_dict, "method", "Invalid prepare response from ", res_rt.resource_uri,
". Output view ", out_name).upper()
if out.http_method != val:
raise SnapValueError("Prepare response for resource %s (%s) has method %s for output view \"%s\", "
"when http method expected was %s" %
(res_rt.resource_uri, res_rt.resource_name, val, out_name, out.http_method))
out.uri = get_non_null_value(out_dict, "uri", "Invalid prepare response from ", res_rt.resource_uri,
". Output view ", out_name)
mode = get_non_null_value(out_dict, "mode", "Invalid prepare response from ", res_rt.resource_uri,
". Output view ", out_name).lower()
if mode == "record":
out.is_record = True
elif mode == "binary":
out.is_record = False
else:
raise SnapValueError("Prepare response for resource %s (%s) has mode %s for output view \"%s\" " %
(res_rt.resource_uri, res_rt.resource_name, mode, out_name))
def create_start_request(notification_uri = None):
"""
Create a START request. This start request can be optionally specify a notification URI. This URI will be
used by the recipient of START request to notify, when the started resource has completed.
{
"snap_control_version": "1.0",
"state": "start"
# notification URI is optional.
"notification_uri": "http://__snap__/runtime/status/...."
}
@return: Message to be sent as START request.
@rtype: dict
"""
mesg = init_message()
mesg["state"] = "start"
if notification_uri is not None:
mesg["notification_uri"] = notification_uri
return mesg
def create_stop_request():
"""
Create a STOP request.
{
"snap_control_version": "1.0",
"state": "stop"
}
@return: Message to be sent as STOP request.
@rtype: dict
"""
mesg = init_message()
mesg["state"] = "stop"
return mesg
def parse_put_request(mesg):
"""
Parse PUT request and see if it is a start, stop or notification request.
@param mesg: The SnapControl message in the PUT request.
@type mesg: dict
@return: True, if it is a start request, False, if it is a stop request.
@rtype: bool
"""
if "state" not in mesg or not mesg["state"]:
if "resource" in mesg:
# This is a notification request.
resource_status = mesg["resource"]
return ("notification", resource_status)
else:
raise SnapValueError("PUT request to runtime status URI is invalid")
elif mesg["state"].lower() == "start":
notify_uri = None
if "notification_uri" in mesg and mesg["notification_uri"] is not None:
notify_uri = mesg["notification_uri"]
return ("start", notify_uri)
elif mesg["state"].lower() == "stop":
return ("stop", None)
else:
raise SnapValueError("PUT request to runtime status URI had unknown state value \"%s\"" % mesg["state"])
def create_notification_request(resource_name, status_obj):
"""
Create a notification request which sends a resource's finished status.
This is sent when resource has finished and wants to notify the pipeline of its finished state
and statistics.
"""
mesg = init_message()
mesg["resource"] = {}
mesg["resource"][resource_name] = {}
_create_status_dict(mesg["resource"][resource_name], status_obj)
return mesg
def parse_notification_request(res_dict):
(resource_name, status_dict) = res_dict.popitem()
status_obj = parse_status_response(status_dict, False)
return (resource_name, status_obj)
def create_start_stop_response(status_obj):
"""
Create a response for a START or STOP request.
{
"snap_control_version": "1.0",
"state_with_time_stamp" : ["Started", 1175287080]
}
@param status_obj: The status object that holds the state information that should be sent back to
the PM or client that made the START or STOP request.
@type status_obj: L{RuntimeStatus}
@return: Message to be sent as the response.
@rtype: dict
"""
mesg = init_message()
mesg["state_with_time_stamp"] = status_obj.get_state_with_ts()
return mesg
def parse_start_stop_response(mesg, status_obj):
"""
Parse the response to a START/STOP request.
@param mesg: The message received as response.
@type mesg: dict
@param status_obj: The status object that needs to be updated with state information in the response.
@type status_obj: L{RuntimeStatus}
"""
state_ts = get_non_null_value(mesg, "state_with_time_stamp", "Status response invalid")
status_obj.set_state_with_ts(state_ts[0], state_ts[1])
def create_status_info(status_obj, resources = None):
"""
Create response message for a status GET request.
The format of status response is along these lines:
{
"snap_control_version" : "1.0",
"state_with_time_stamp" : ["Started", 1175287080]
"creation_time_stamp" : 1175287050, # IN UTC format....might be changed to datetime type
"input_views" : {
"inp1" : {"record_count" : 1000},
"inp2" : {"record_count" : 250}
},
"output_views" : {
"out1" : {"record_count" : 1000},
"out2" : {"record_count" : 250}
},
# This sub dictionary only appears if the level is set to detail and the resource is pipeline
# with resources inside it.
"resources" : {
"reader" : {
'state_with_time_stamp' : ['Running', 1175287080],
'creation_time_stamp' : "....",
"input_views" : {
"inp1" : {"record_count" : 1000}
},
"output_views" : {
"out1" : {"record_count" : 1000}
}
},
"writer" : {
#....
}
}
}
@param status_obj: The status object that is walked to fetch status information
@type status_obj: L{RuntimeStatus}
@param resources: Dictionary with name of resource in the pipeline as key and status object of that
resource as value. This param is only needed when detailed status has been requested. Detailed
status includes status of not just the pipeline, but also of the resources inside the pipeline.
@type resources: dict or None
@param is_detailed: If set to True, then status information of resources inside the pipeline
are also fetched. This param is useful onloy when the status request is for pipeline resource
@type is_detailed: bool
@return: Message to be sent to status requestor.
@rtype: dict
"""
mesg = init_message()
_create_status_dict(mesg, status_obj, resources)
return mesg
def _create_status_dict(mesg, status_obj, resources = None):
"""
Populate message dictionary with status object information.
@param mesg: Message dictionary that must be populated.
@type mesg: dict
@param status_obj: The status object of the resource or pipeline
@type status_obj: L{RuntimeStatus}
@param resources: Dictionary with name of resource in the pipeline as key and status object of that
resource as value. This param is only needed when detailed status has been requested and that
too from a pipeline resource. Detailed status includes status of not just the pipeline, but also
that of the resources inside the pipeline.
@type resources: dict or None
"""
mesg["state_with_time_stamp"] = status_obj.get_state_with_ts()
mesg["creation_time_stamp"] = status_obj.create_ts
mesg["statistics"] = copy.deepcopy(status_obj.statistics)
if resources is not None:
mesg["resources"] = {}
for name in resources:
mesg["resources"][name] = {}
_create_status_dict(mesg["resources"][name], resources[name])
def parse_status_response(mesg, is_detailed):
"""
Parse the status response from a resource runtime and create status object(s) from it.
@param mesg: The message received as status response.
@type mesg: dict
@param is_detailed: If set to True, then the response can have status information of
resources inside the pipeline.
@type is_detailed: bool
@return: Returns the status object
@rtype: L{RuntimeStatus}
"""
status_obj = RuntimeStatus()
state_ts = get_non_null_value(mesg, "state_with_time_stamp", "Status response invalid")
status_obj.set_state_with_ts(state_ts[0], state_ts[1])
ts = get_non_null_value(mesg, "creation_time_stamp", "Status response invalid")
status_obj.create_ts = ts
status_obj.statistics = mesg["statistics"]
if is_detailed and "resources" in mesg:
for r in mesg["resources"]:
stats = parse_status_response(mesg["resources"][r], False)
status_obj["resources"][r] = stats
return status_obj
|