initial
diff --git a/examples/aria-service-proxy-plugin/.gitignore b/examples/aria-service-proxy-plugin/.gitignore
index e26c3bb..20b16b2 100644
--- a/examples/aria-service-proxy-plugin/.gitignore
+++ b/examples/aria-service-proxy-plugin/.gitignore
@@ -3,6 +3,7 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
+.travis.yml
# C extensions
*.so
diff --git a/examples/aria-service-proxy-plugin/.travis.yml b/examples/aria-service-proxy-plugin/.travis.yml
deleted file mode 100644
index ba05d7a..0000000
--- a/examples/aria-service-proxy-plugin/.travis.yml
+++ /dev/null
@@ -1,12 +0,0 @@
-dist: trusty
-
-language: python
-python:
- - '2.7'
-
-install:
- - pip install -r requirements.txt
-
-script:
- - pytest
-
diff --git a/examples/aria-service-proxy-plugin/README.md b/examples/aria-service-proxy-plugin/README.md
index ad50a99..e97cf1a 100644
--- a/examples/aria-service-proxy-plugin/README.md
+++ b/examples/aria-service-proxy-plugin/README.md
@@ -5,15 +5,15 @@
## Types
### aria.serviceproxy.ServiceProxy
-Abstract representation of an ARIA service. The type operates by copying the outputs of target services into its local runtime attributes. The service must be installed, and node properties control other conditions for successful instantiation. It isn't assumed that the service exists at the time of instantiation, and properties control timeout behavior. A failed service discovery, or listed desired outputs, results in a NonRecoverableError.
+Abstract representation of an ARIA service. The type operates by copying the outputs of target services into its local runtime attributes. The service must be installed, and node properties control other conditions for successful instantiation. It isn't assumed that the service exists at the time of instantiation, and properties control timeout behavior. A failed service discovery, or listed desired outputs, results in a NonRecoverableError. The plugin logic is entirely executed in the `create` lifecycle operation of the Standard interface. Any template nodes dependent on the proxy will wait for the proxy to either gather the outputs of the target service, or timeout, which will abort the workflow. The intended usage is for dependent nodes to have easy access to the outputs of ARIA managed services via attributes.
#### Properties
-* __service_name__ : The name of the target service.
-* __outputs__ : A `string` list of service outputs. Acts as a filter to selected desired outputs. An empty list will result in no outputs being copied.
+* __service_name__ : The name of the target service. Required.
+* __outputs__ : A `string` list of service outputs. Acts as a filter to selected desired outputs. An empty list will result in no outputs being copied, which also would have the effect of making the existence of the target sufficient to satisfy the proxy.
* __wait_config__: A `dictionary` with two keys:
* * __wait_for_service__: A `boolean` that indicates whether to wait for a service that doesn't exist yet. If `False`, if either the service or requested outputs are not present immediately, a non-recoverable exception is thrown. If `True`, the plugin will wait.
-* * __wait_time__: An `integer` that indicates the number of seconds to wait for the service, and specified outputs, to exist. Beyond this time, a non-recoverable exception is thrown.
-* * __wait_expression__: A Python evaluatable boolean expression using output names. The expression acts as another barrier to matching the target service. Examples values: "output1 > 4", "len(output1) == 7", "status == "ready"".
+* * __wait_time__: An `integer` that indicates the number of seconds to wait for the service, and specified outputs, to exist. Beyond this time, a non-recoverable exception is thrown. Note that there is no enforced maximum wait which could cause a misconfiguration to wait forever.
+* * __wait_expression__: A Python evaluatable boolean expression using output names. The expression acts as another barrier to matching the target service. Examples values: "output1 > 4", "len(output1) == 7", "status == "ready"". Be aware that while syntax is checked via Python evaluation, the meaning of the expression is not checked. For example a wait expression of "1==0" could be supplied and result in an unsatifiable condition and wait (or failure). Note that the Python evaluation environment only contains the target service outputs, and no additional imports.
#### Attributes
-The outputs of the target are placed in a `list` in the runtime attributes named `service_outputs`. Service outputs entries consist of single entry dictionaries, with the keys `name` and `value`.
+The outputs of the target are copied in a `list` in the runtime attributes named `service_outputs`. Service outputs entries consist of single entry dictionaries, with the keys `name` and `value`. Service output names and values are unchanged from their original values in the proxied service.
diff --git a/examples/aria-service-proxy-plugin/aria_service_proxy/tasks.py b/examples/aria-service-proxy-plugin/aria_service_proxy/tasks.py
index 2470d64..6a17bb1 100644
--- a/examples/aria-service-proxy-plugin/aria_service_proxy/tasks.py
+++ b/examples/aria-service-proxy-plugin/aria_service_proxy/tasks.py
@@ -19,61 +19,91 @@
from datetime import datetime
from aria.cli.core import aria
+WAIT_CONFIG_KEY = 'wait_config'
+WAIT_TIME_KEY = 'wait_time'
+WAIT_FOR_SERVICE_KEY = 'wait_for_service'
+WAIT_EXPR_KEY = 'wait_expression'
+SERVICE_NAME_KEY = 'service_name'
+SERVICE_OUTPUTS_KEY = 'service_outputs'
+OUTPUT_KEY = 'outputs'
+LAST_UPDATE_KEY = 'last_update'
+
+WORKFLOW_INSTALL = 'install'
+WORKFLOW_UNINSTALL = 'uninstall'
+WF_SUCCESS_STATUS = 'succeeded'
+
+RETRY_DELAY_SECS = 1
+
+# Only operation exposed. Waits for proxied service to be ready
+# if necessary, then copies selected outputs to node attributes.
+#
@operation
def proxy_connect(**kwargs):
+ #
+ # Collect node configuration
+ #
+
service_names = get_service_names()
- duration = ctx.node.properties['wait_config']['wait_time']
+ duration = ctx.node.properties[WAIT_CONFIG_KEY][WAIT_TIME_KEY]
installed = False
- service_exists = ctx.node.properties['service_name'] in service_names
+ service_exists = ctx.node.properties[SERVICE_NAME_KEY] in service_names
if service_exists:
installed = is_installed(
- service_names[ctx.node.properties['service_name']])
+ service_names[ctx.node.properties[SERVICE_NAME_KEY]])
- wait_for_service = ctx.node.properties['wait_config']['wait_for_service']
+ wait_for_service = ctx.node.properties[WAIT_CONFIG_KEY][WAIT_FOR_SERVICE_KEY]
- wait_expr = ctx.node.properties['wait_config'].get('wait_expression',
+ wait_expr = ctx.node.properties[WAIT_CONFIG_KEY].get(WAIT_EXPR_KEY,
None)
+ # If the service doesn't exist, or exists but hasn't been installed,
+ # and wait_for_service = true, retry for configured duration.
+ #
if not service_exists or not installed:
if wait_for_service:
if duration > ctx.operation.retry_number:
return ctx.operation.retry(
- message = 'Waiting for service', retry_after=1)
+ message = 'Waiting for service',
+ retry_after=RETRY_DELAY_SECS)
else:
if not service_exists:
raise NonRecoverableError(
"service {} not found".format(
- ctx.node.properties['service_name']))
+ ctx.node.properties[SERVICE_NAME_KEY]))
else:
raise NonRecoverableError(
"service {} not installed".format(
- ctx.node.properties['service_name']))
+ ctx.node.properties[SERVICE_NAME_KEY]))
else:
if not service_exists:
raise NonRecoverableError(
"service {} not found".format(
- ctx.node.properties['service_name']))
+ ctx.node.properties[SERVICE_NAME_KEY]))
else:
raise NonRecoverableError(
"service {} not installed".format(
- ctx.node.properties['service_name']))
+ ctx.node.properties[SERVICE_NAME_KEY]))
- # Service exists, so see if outputs exist yet
- elif( ctx.node.properties['outputs']):
- outputs = service_names[ctx.node.properties['service_name']].outputs
- if not output_equivalence( ctx.node.properties['outputs'], outputs):
+ # Service ready. If outputs are configured in proxy, grab them
+ elif( ctx.node.properties[OUTPUT_KEY]):
+ outputs = service_names[ctx.node.properties[SERVICE_NAME_KEY]].outputs
+
+ # If the outputs are not ready yet, retry
+ if not output_equivalence( ctx.node.properties[OUTPUT_KEY], outputs):
return fail_or_wait(wait_for_service,
duration,
'Waiting for service outputs',
"service {} outputs {} not found".format(
- ctx.node.properties['service_name'],
- ctx.node.properties['outputs']))
+ ctx.node.properties[SERVICE_NAME_KEY],
+ ctx.node.properties[OUTPUT_KEY]))
+ # Outputs are ready. Copy them from target outputs into the
+ # this node instance attributes
else:
# Success
# place outputs in attributes
@@ -86,15 +116,17 @@
wait_expr)))
service_outputs = []
- if('service_outputs' in ctx.instance.runtime_properties and
- ctx.instance.runtime_properties['service_outputs']):
+ if(SERVICE_OUTPUTS_KEY in ctx.instance.runtime_properties and
+ ctx.instance.runtime_properties[SERVICE_OUTPUTS_KEY]):
service_outputs = list(
- ctx.instance.runtime_properties['service_outputs'])
- for k,v in outputs.iteritems():
- service_outputs.append( dict(name = k,value = v.value))
- ctx.instance.runtime_properties['service_outputs'] = service_outputs
+ ctx.instance.runtime_properties[SERVICE_OUTPUTS_KEY])
+ for key,val in outputs.iteritems():
+ service_outputs.append( dict(name = key,value = val.value))
+ ctx.instance.runtime_properties[SERVICE_OUTPUTS_KEY] = service_outputs
- ctx.instance.runtime_properties['last_update'] = str(datetime.utcnow())
+ ctx.instance.runtime_properties[LAST_UPDATE_KEY] = str(datetime.utcnow())
+
+ # Service exists, but outputs not configured, so we're done
else:
ctx.logger.info("service exists, but no outputs specified = success")
@@ -112,43 +144,48 @@
return outdict
-# Tests whether the list of configured outputs (a simple string list) is equivalent
-# to the list returned from Aria (possible duplicate keys)
-def output_equivalence( config_list, service_list ):
+# Tests whether the list of configured outputs (a simple string list)
+# is equivalent # to the list returned from Aria (possible duplicate keys)
+def output_equivalence(config_list, service_list):
sset = set()
- for k,v in service_list.iteritems():
- sset.add( k )
+ for key, val in service_list.iteritems():
+ sset.add(key)
if not len(sset) == len(config_list):
return False
for entry in sset:
- if not entry in config_list:
+ if entry not in config_list:
return False
return True
+
# Looks at the execution history to determine of service is installed
@aria.pass_model_storage
-def is_installed( service, model_storage ):
+def is_installed(service, model_storage):
executions = model_storage.execution.list(
filters=dict(service=service)).items
- for e in reversed(executions):
- if e.workflow_name == "uninstall":
+ for execution in reversed(executions):
+ if execution.workflow_name == WORKFLOW_UNINSTALL:
return False
- if e.workflow_name == "install" and e.status == "succeeded":
+ if (execution.workflow_name == WORKFLOW_INSTALL and
+ execution.status == WF_SUCCESS_STATUS):
return True
return False
-# Evaluates wait_expr in the context of supplied outputs
-def eval_waitexpr( expr, outputs):
- locals = {}
- for k,v in outputs.iteritems():
- locals[k] = v.value
- return eval(expr,locals)
+# Evaluates wait_expr in the context of supplied outputs
+def eval_waitexpr(expr, outputs):
+ locals = {}
+ for key, val in outputs.iteritems():
+ locals[key] = val.value
+ return eval(expr, locals)
+
+# Convenience function that either fails immediately (if wait_flag
+# is false), or tests and retries or fails based on the wait condition
def fail_or_wait( wait_flag, duration, wait_msg, fail_msg ):
if wait_flag:
if duration > ctx.operation.retry_number:
return ctx.operation.retry(
- message = wait_msg, retry_after=1)
+ message = wait_msg, retry_after=RETRY_DELAY_SECS)
else:
raise NonRecoverableError( fail_msg)
else:
diff --git a/examples/aria-service-proxy-plugin/plugin.yaml b/examples/aria-service-proxy-plugin/plugin.yaml
index 3c3ab90..ee27b19 100644
--- a/examples/aria-service-proxy-plugin/plugin.yaml
+++ b/examples/aria-service-proxy-plugin/plugin.yaml
@@ -39,7 +39,8 @@
default: 30
wait_expression:
description: >-
- boolean expression to wait for truth value. Python syntax.
+ boolean expression to wait for truth value. Python syntax. Outputs from target
+ service are provided as variable names in the expression environment.
Example: output1 > output2
type: string
required: false
@@ -56,7 +57,7 @@
outputs:
description: >-
list of outputs that will be mapped to proxy attributes
- under the \"proxied\" key
+ under the \"service_outputs" key
required: false
default: []
type: list
@@ -64,6 +65,8 @@
type: string
wait_config:
type: serviceproxy.waitconfig
+ description: >-
+ configuration of wait for service existence, outputs, and/or output values
required: false
default:
wait_for_service:
diff --git a/examples/aria-service-proxy-plugin/tests/test_proxy.py b/examples/aria-service-proxy-plugin/tests/test_proxy.py
index 9656a3e..6328b54 100644
--- a/examples/aria-service-proxy-plugin/tests/test_proxy.py
+++ b/examples/aria-service-proxy-plugin/tests/test_proxy.py
@@ -22,21 +22,27 @@
from cloudify.exceptions import NonRecoverableError
from cloudify.state import current_ctx
+
+# Tests that a retry is performed when wait_for_service is true and
+# the service doesn't exist
def test_noservice_with_wait(monkeypatch):
- node = Mock(properties = {'service_name': 'testwait',
- 'wait_config':{'wait_for_service': True, 'wait_time':5},
- 'outputs': ['someoutput']})
- attrs = {'retry.return_value' : 'retry' }
- operation = Mock( retry_number = 0, **attrs)
- ctxmock = MagicMock( node = node , operation = operation)
+ node = Mock(properties={'service_name': 'testwait',
+ 'wait_config':
+ {'wait_for_service': True, 'wait_time': 5},
+ 'outputs': ['someoutput']})
+ attrs = {'retry.return_value': 'retry'}
+ operation = Mock(retry_number=0, **attrs)
+ ctxmock = MagicMock(node = node, operation = operation)
current_ctx.set(ctxmock)
monkeypatch.setattr(tasks,"get_service_names",lambda: {})
ret = tasks.proxy_connect()
assert(ret == 'retry')
+
+# Make sure error raised when no service exists and no wait is configured
def test_noservice_no_wait(monkeypatch):
- node = Mock(properties = {'service_name': 'testfail',
- 'wait_config':{'wait_for_service': False, 'wait_time':5 },
+ node = Mock(properties={'service_name': 'testfail',
+ 'wait_config': {'wait_for_service': False, 'wait_time': 5 },
'outputs': ['someoutput']})
ctxmock = MagicMock(node = node)
current_ctx.set(ctxmock)
@@ -44,6 +50,10 @@
with pytest.raises(NonRecoverableError):
tasks.proxy_connect()
+
+# Test that a service for which wait is specified, is retried
+# properly, and the retry results in success after the service
+# appears
def test_service_eventual_complete(monkeypatch):
output = Mock( value = 'someval')
service = Mock(outputs = { 'someoutput': output })
@@ -75,11 +85,15 @@
assert(ctxmock.instance.runtime_properties['service_outputs'][0]['name'] == 'someoutput')
assert(ctxmock.instance.runtime_properties['service_outputs'][0]['value'] == 'someval')
+
+# Test that a proxy generates a retry when the service exists, but the
+# outputs don't, and wait is configured
def test_output_retry(monkeypatch):
output = Mock( value = 'someval')
service = Mock(outputs= {'test':output})
node = Mock(properties = {'service_name': 'test',
- 'wait_config':{'wait_for_service': True, 'wait_time':5},
+ 'wait_config':
+ {'wait_for_service': True, 'wait_time':5},
'outputs': ['someoutput']})
attrs = {'retry.return_value' : 'retry' }
operation = Mock( retry_number = 0, **attrs)
@@ -91,6 +105,8 @@
ret = tasks.proxy_connect()
assert(ret == 'retry')
+
+# Test that nominal path works: service complete, outputs available
def test_output_complete(monkeypatch):
output = Mock( value = 'someval')
service = Mock(outputs= {'someoutput':output})
@@ -107,6 +123,8 @@
ret = tasks.proxy_connect()
assert(ret == None)
+# Test that retry occurs propertly when outputs unavailable (and wait
+# configured), and that they are propertly detected when they appear.
def test_output_eventual_complete(monkeypatch):
output = Mock( value = 'someval')
service = Mock(outputs= {'test':output})
@@ -132,6 +150,8 @@
assert(ctxmock.instance.runtime_properties['service_outputs'][0]['name'] == 'someoutput')
assert(ctxmock.instance.runtime_properties['service_outputs'][0]['value'] == 'someval')
+# Test wait expression fuctionality for failure case. Pass in an output
+# with and test for incorrect length. Should raise error.
def test_expr_fail(monkeypatch):
output = Mock( value = 'someval')
service = Mock(outputs= {'someoutput':output})
@@ -151,6 +171,8 @@
with pytest.raises(NonRecoverableError):
tasks.proxy_connect()
+# Test wait expression fuctionality for success case. Pass in an output
+# with and test for correct length. Should raise no error.
def test_expr_succeed(monkeypatch):
output = Mock( value = 'someval')
service = Mock(outputs= {'someoutput':output})
@@ -170,6 +192,9 @@
ret = tasks.proxy_connect()
assert(ret == None)
+# Test wait expression fuctionality for success case using two outputs.
+# Expression is boolean expression involving two outputs, and it true.
+# Should raise no error.
def test_expr_succeed_mult(monkeypatch):
output1 = Mock( value = 4)
output2 = Mock( value = 7)
