Index: fedd/federation/experiment_control.py
===================================================================
--- fedd/federation/experiment_control.py	(revision 57facaedaf06eed2280a5c0b375da51f5208b327)
+++ fedd/federation/experiment_control.py	(revision 22a1a77015d7254a2c8a9f81b420a91df69516f5)
@@ -37,4 +37,5 @@
 from thread_pool import thread_pool, pooled_thread
 from experiment_info import experiment_info, allocation_info, federated_service
+from operation_status import operation_status
 
 import topdl
@@ -65,4 +66,5 @@
     call_TerminateSegment = service_caller('TerminateSegment')
     call_InfoSegment = service_caller('InfoSegment')
+    call_OperationSegment = service_caller('OperationSegment')
     call_Ns2Topdl = service_caller('Ns2Topdl')
 
@@ -223,4 +225,5 @@
                 'Info': soap_handler('Info', self.get_info),
                 'MultiInfo': soap_handler('MultiInfo', self.get_multi_info),
+                'Operation': soap_handler('Operation', self.do_operation),
                 'Terminate': soap_handler('Terminate', 
                     self.terminate_experiment),
@@ -238,4 +241,5 @@
                 'Terminate': xmlrpc_handler('Terminate',
                     self.terminate_experiment),
+                'Operation': xmlrpc_handler('Operation', self.do_operation),
 		'GetValue': xmlrpc_handler('GetValue', self.GetValue),
 		'SetValue': xmlrpc_handler('SetValue', self.SetValue),
@@ -933,4 +937,43 @@
 	    except service_error, e:
 		self.log.error("Info segment failed on %s: %s" % \
+			(self.testbed, e))
+		return False
+
+    class operation_segment:
+	def __init__(self, debug=False, log=None, testbed="", cert_file=None,
+		cert_pwd=None, trusted_certs=None, caller=None,
+		log_collector=None):
+	    self.log = log
+	    self.debug = debug
+	    self.cert_file = cert_file
+	    self.cert_pwd = cert_pwd
+	    self.trusted_certs = None
+	    self.caller = caller
+	    self.testbed = testbed
+	    self.status = None
+
+	def __call__(self, uri, aid, op, targets, params):
+	    req = { 
+		    'allocID': { 'fedid' : aid },
+		    'operation': op,
+		    'target': targets,
+		    }
+	    if params: req['parameter'] = params
+
+
+	    try:
+		self.log.debug("Calling OperationSegment at %s " % uri)
+		r = self.caller(uri, req, self.cert_file, self.cert_pwd,
+			self.trusted_certs)
+		if 'OperationSegmentResponseBody' in r:
+		    r = r['OperationSegmentResponseBody']
+		    if 'status' in r:
+			self.status = r['status']
+		else:
+		    raise service_error(service_error.internal, 
+			    "Bad response!?: %s" %r)
+		return True
+	    except service_error, e:
+		self.log.error("Operation segment failed on %s: %s" % \
 			(self.testbed, e))
 		return False
@@ -2314,4 +2357,120 @@
             raise service_error(service_error.req, "No such experiment")
 
+    def operate_on_segments(self, op_params, cert, op, testbeds, params, 
+	    results):
+	"""
+	Call OperateSegment on multiple testbeds and gather the results.
+	op_params contains the parameters needed to contact that testbed, cert
+	is a certificate containing the fedid to use, op is the operation,
+	testbeds is a dict mapping testbed name to targets in that testbed,
+	params are the parameters to include a,d results is a growing list of
+	the results of the calls.
+	"""
+	try:
+	    tmpdir = tempfile.mkdtemp(prefix="info-")
+	except EnvironmentError:
+	    raise service_error(service_error.internal, 
+		    "Cannot create tmp dir")
+	cert_file = self.make_temp_certfile(cert, tmpdir)
+
+	try:
+	    for tb, targets in testbeds.items():
+		if tb in op_params:
+		    uri, aid = op_params[tb]
+		    operate=self.operation_segment(log=self.log, testbed=uri,
+				cert_file=cert_file, cert_pwd=None,
+				trusted_certs=self.trusted_certs,
+				caller=self.call_OperationSegment)
+		    if operate(uri, aid, op, targets, params):
+			if operate.status is not None:
+			    results.extend(operate.status)
+			    continue
+		# Something went wrong in a weird way.  Add statuses 
+		# that reflect that to results
+		for t in targets:
+		    results.append(operation_status(t, 
+			operation_status.federant,
+			'Unexpected error ion %s' % tb))
+	# Clean up the tmpdir no matter what
+	finally:
+	    if tmpdir: self.remove_dirs(tmpdir)
+
+    def do_operation(self, req, fid):
+	"""
+	Find the testbeds holding each target and ask them to carry out the
+	operation.  Return the statuses.
+	"""
+	# Map an element to the testbed containing it
+	def element_to_tb(e):
+	    if isinstance(e, topdl.Computer): return e.get_attribute("testbed")
+	    elif isinstance(e, topdl.Testbed): return e.name
+	    else: return None
+	# If d is an operation_status object, make it a dict
+	def make_dict(d):
+	    if isinstance(d, dict): return d
+	    elif isinstance(d, operation_status): return d.to_dict()
+	    else: return { }
+
+        req = req.get('OperationRequestBody', None)
+        if not req:
+            raise service_error(service_error.req,
+                    "Bad request format (no OperationRequestBody)")
+        exp = req.get('experiment', None)
+	op = req.get('operation', None)
+	target = set(req.get('target', []))
+	params = req.get('parameter', None)
+
+        if exp:
+            if 'fedid' in exp:
+                key = exp['fedid']
+                keytype = "fedid"
+            elif 'localname' in exp:
+                key = exp['localname']
+                keytype = "localname"
+            else:
+                raise service_error(service_error.req, "Unknown lookup type")
+        else:
+            raise service_error(service_error.req, "No request?")
+
+	if op is None or not target
+            raise service_error(service_error.req, "No request?")
+
+        proof = self.check_experiment_access(fid, key)
+	self.state_lock.acquire()
+	if key in self.state:
+	    d1, op_params, cert, d2 = \
+		    self.get_segment_info(self.state[key], need_lock=False)
+	    top = self.state[key].top
+	    if top is not None:
+		top = top.clone()
+	self.state_lock.release()
+
+	if top is None:
+	    raise service_error(service_error.partial, "No topology yet", 
+		    proof=proof)
+
+	testbeds = { }
+	results = []
+	for e in top.elements:
+	    if e.name in targets:
+		tb = element_to_tb(e)
+		if tb is not None:
+		    if tb in testbeds: testbeds[tb].append(e.name)
+		    else: testbeds[tb] = [ e.name ]
+		else:
+		    results.append(operation_status(e.name, 
+			code=operation_status.no_target, 
+			description='Cannot map target to testbed'))
+
+	self.operate_on_segments(op_params, cert, op, testbeds, params,
+		results)
+
+	return { 
+		'experiment': exp, 
+		'status': [make_dict(r) for r in results]
+		'proof': proof.to_dict()
+		}
+
+
     def get_multi_info(self, req, fid):
         """
