def execute_query(self, discovery_query, id_only=True, query_args=None, query_params=None):
        try:
            if "QUERYEXP" in discovery_query:
                ds_query, ds_name = discovery_query, discovery_query["query_args"].get("datastore", DataStore.DS_RESOURCES)
            else:
                log.info("DatastoreDiscovery.execute_query(): discovery_query=\n%s", pprint.pformat(discovery_query))
                ds_query, ds_name = self._build_ds_query(discovery_query, id_only=id_only)

            current_actor_id=get_ion_actor_id(self.process)
            ds_query.setdefault("query_params", {})
            if query_params:
                ds_query["query_params"].update(query_params)
            ds_query["query_params"]["current_actor"] = current_actor_id

            log.debug("DatastoreDiscovery.execute_query(): ds_query=\n%s", pprint.pformat(ds_query))

            ds = self._get_datastore(ds_name)
            access_args = create_access_args(current_actor_id=current_actor_id,
                                             superuser_actor_ids=self.container.resource_registry.get_superuser_actors())
            query_results = ds.find_by_query(ds_query, access_args=access_args)
            log.info("Datastore discovery query resulted in %s rows", len(query_results))

            if query_args and query_args.get("query_info", False):
                query_info = dict(_query_info=True, query=ds_query, access_args=access_args, ds_name=ds_name)
                query_info.update(ds_query.get("_result", {}))
                query_results.append(query_info)

            return query_results
        except Exception as ex:
            log.exception("DatastoreDiscovery.execute_query() failed")
        return []

    def tearDown(self):
        new_policy = {
            'metric': 'app_attributes:ml',
            'sample_period': 600,
            'sample_function': 'Average',
            'cooldown_period': 0,
            'scale_up_threshold': 2.0,
            'scale_up_n_processes': 1,
            'scale_down_threshold': 1.0,
            'scale_down_n_processes': 1,
            'maximum_processes': 0,
            'minimum_processes': 0,
        }
        self.haa_client.reconfigure_policy(new_policy)

        self.waiter.await_state_event(state=ProcessStateEnum.TERMINATED)
        self.assertEqual(len(self.get_running_procs()), 0)

        self.waiter.stop()
        # Going in for an extra kill if pthread is stilling running
        ha_proc = self.container.proc_manager.procs.get(self._haa_pid, None)
        pthread = None
        if ha_proc:
            pthread = ha_proc.policy_thread
        try:
            self.container.terminate_process(self._haa_pid)
        except BadRequest:
            log.exception("Couldn't terminate HA Agent in teardown (May have been terminated by a test)")
            raise
        finally:
            if pthread:
                pthread.kill()
            self._stop_webserver()
            self._stop_container()

def get_version_info():
    import pkg_resources
    pkg_list = ["coi-services",
                "pyon",
                "coverage-model",
                "ion-functions",
                "eeagent",
                "epu",
                "utilities",
                "marine-integrations"]

    version = {}
    for package in pkg_list:
        try:
            version["%s-release" % package] = pkg_resources.require(package)[0].version
            # @TODO git versions for each?
        except pkg_resources.DistributionNotFound:
            pass

    try:
        dir_client = DirectoryServiceProcessClient(process=service_gateway_instance)
        sys_attrs = dir_client.lookup("/System")
        if sys_attrs and isinstance(sys_attrs, dict):
            version.update({k: v for (k, v) in sys_attrs.iteritems() if "version" in k.lower()})
    except Exception as ex:
        log.exception("Could not determine system directory attributes")

    return gateway_json_response(version)

    def run_mission(self, mission_id, mission_loader, mission_scheduler, instrument_objs):
        """
        Runs a mission returning to caller when the execution is completed.
        Parameters as returned by load_mission.
        """

        if mission_id in self._running_missions:
            raise BadRequest('run_mission: mission_id=%r is already running', mission_id)

        self._running_missions[mission_id] = mission_scheduler
        log.debug('%r: [mm] starting mission_id=%r (#running missions=%s)', self._platform_id,
                  mission_id, len(self._running_missions))
        try:
            mission_scheduler.run_mission()
        except Exception as ex:
            log.exception('%r: [mm] run_mission mission_id=%r', self._platform_id, mission_id)
        finally:
            del self._running_missions[mission_id]

            # remove exclusive access:
            mission_entries = mission_loader.mission_entries
            for mission_entry in mission_entries:
                instrument_ids = mission_entry.get('instrument_id', [])
                for instrument_id in instrument_ids:
                    if instrument_id in instrument_objs:
                        resource_id = instrument_objs[instrument_id].resource_id
                        self._remove_exclusive_access(instrument_id, resource_id, mission_id)

            log.debug('%r: [mm] completed mission_id=%r (#running missions=%s)', self._platform_id,
                      mission_id, len(self._running_missions))

    def load_mission(self, mission_id, mission_yml):
        """
        Loads a mission as preparation prior to its actual execution.

        @param mission_id
        @param mission_yml
        @return (mission_loader, mission_scheduler, instrument_objs) arguments
                for subsequence call to run_mission
        @raise BadRequest if mission_id is already running or there's any
                          problem loading the mission
        """

        if mission_id in self._running_missions:
            raise BadRequest('run_mission: mission_id=%r is already running', mission_id)

        try:
            mission_loader, mission_scheduler, instrument_objs = \
                self._create_mission_scheduler(mission_id, mission_yml)
        except Exception as ex:
            msg = '%r: [mm] run_mission: mission_id=%r _create_mission_scheduler exception: %s' % (
                self._platform_id, mission_id, ex)
            log.exception(msg)
            raise BadRequest(msg)

        return mission_id, mission_loader, mission_scheduler, instrument_objs

    def _call_plugins(self, method, process, config, **kwargs):
        bootstrap_plugins = config.get_safe("bootstrap_plugins", None)
        if bootstrap_plugins is None:
            log.warn("Bootstrapper called without bootstrap_plugins config")

        # Finding the system actor ID. If found, construct call context headers.
        # This may be called very early in bootstrap with no system actor yet existing
        system_actor, _ = process.container.resource_registry.find_resources(
            RT.ActorIdentity, name=self.CFG.system.system_actor, id_only=True
        )
        system_actor_id = system_actor[0] if system_actor else "anonymous"

        actor_headers = {
            "ion-actor-id": system_actor_id,
            "ion-actor-roles": {"ION": ["ION_MANAGER", "ORG_MANAGER"]} if system_actor else {},
        }

        # Set the call context of the current process
        with process.push_context(actor_headers):

            for plugin_info in bootstrap_plugins:
                plugin_mod, plugin_cls = plugin_info.get("plugin", [None, None])
                plugin_cfg = plugin_info.get("config", None)
                plugin_cfg = dict_merge(config, plugin_cfg) if plugin_cfg is not None else config

                try:
                    log.info("Bootstrapping plugin %s.%s ...", plugin_mod, plugin_cls)
                    plugin = for_name(plugin_mod, plugin_cls)
                    plugin_func = getattr(plugin, method)
                    plugin_func(process, plugin_cfg, **kwargs)
                except AbortBootstrap as abort:
                    raise
                except Exception as ex:
                    log.exception("Error bootstrapping plugin %s.%s", plugin_mod, plugin_cls)

    def _call_plugins(self, method, process, config, **kwargs):
        bootstrap_plugins = config.get_safe("bootstrap_plugins", None)
        if bootstrap_plugins is None:
            log.warn("Bootstrapper called without bootstrap_plugins config")

        # Finding the system actor ID. If found, construct call context headers.
        # This may be called very early in bootstrap with no system actor yet existing
        system_actor = get_system_actor()
        if system_actor:
            actor_headers = get_system_actor_header(system_actor)
        else:
            # Use default actor headers, not roles.
            actor_headers = build_actor_header()

        # Set the call context of the current process
        with process.push_context(actor_headers):

            for plugin_info in bootstrap_plugins:
                plugin_mod, plugin_cls = plugin_info.get("plugin", [None,None])
                plugin_cfg = plugin_info.get("config", None)
                plugin_cfg = dict_merge(config, plugin_cfg) if plugin_cfg is not None else config

                try:
                    log.info("Bootstrapping plugin %s.%s ...", plugin_mod, plugin_cls)
                    plugin = for_name(plugin_mod, plugin_cls)
                    plugin_func = getattr(plugin, method)
                    plugin_func(process, plugin_cfg, **kwargs)
                except AbortBootstrap as abort:
                    raise
                except Exception as ex:
                    log.exception("Error bootstrapping plugin %s.%s", plugin_mod, plugin_cls)

    def _device_removed_event(self, evt):
        """
        Handles the device_removed event to remove associated information and
        status updates, which mauy result in events being published.
        """

        # the actual child removed is in the values component of the event:
        if isinstance(evt.values, (list, tuple)):
            # normally it will be just one element but handle as array:
            for sub_resource_id in evt.values:
                self._remove_child(sub_resource_id)
        else:
            log.warn("%r: Got device_removed event with invalid values member: %r",
                     self._platform_id, evt)
            return

        # finally forward event so ancestors also get notified:
        # only adjustment is that now my platform's resource_id is the origin:
        evt = dict(event_type  = evt.type_,
                   sub_type    = evt.sub_type,
                   origin_type = evt.origin_type,
                   origin      = self.resource_id,
                   description = evt.description,
                   values      = evt.values)
        try:
            log.debug('%r: _device_removed_event: forwarding to ancestors: %s',
                      self._platform_id, evt)

            self._event_publisher.publish_event(**evt)

        except Exception:
            log.exception('%r: platform agent could not publish event: %s',
                          self._platform_id, evt)

    def persist_or_timeout(self, stream_id, rdt):
        """ retry writing coverage multiple times and eventually time out """
        done = False
        timeout = 2
        start = time.time()
        while not done:
            try:
                self.add_granule(stream_id, rdt)
                done = True
            except:
                log.exception('An issue with coverage, retrying after a bit')
                if (time.time() - start) > MAX_RETRY_TIME: # After an hour just give up
                    dataset_id = self.get_dataset(stream_id)
                    log.error("We're giving up, the coverage needs to be inspected %s", DatasetManagementService._get_coverage_path(dataset_id))
                    raise

                if stream_id in self._coverages:
                    log.info('Popping coverage for stream %s', stream_id)
                    self._coverages.pop(stream_id)

                gevent.sleep(timeout)
                if timeout > (60 * 5):
                    timeout = 60 * 5
                else:
                    timeout *= 2

    def _finalize_uirefs(self, ds):
        # Create real resource IDs
        for obj in self.ui_objs.values():
            oid = self.uiid_prefix + obj.uirefid
            obj._id = oid
            self.ui_obj_by_id[oid] = obj

            # Change references for all known UI objects
            for attr in obj.__dict__:
                if attr != 'uirefid' and getattr(obj, attr) in self.ui_objs:
                    setattr(obj, attr, self.uiid_prefix + getattr(obj, attr))
            try:
                json.dumps(obj.__dict__.copy())
            except Exception as ex:
                log.exception("Object %s problem" % obj)

        # Resolve associations to real resource IDs
        for refassoc in self.ref_assocs:
            sub_refid, pred, obj_refid = refassoc
            try:
                subo = self.ui_objs[sub_refid]
                objo = self.ui_objs[obj_refid]
                assoc = objects.Association(at="",
                    s=subo._id, st=subo._get_type(), srv="",
                    p=pred,
                    o=objo._id, ot=objo._get_type(), orv="",
                    ts=get_ion_ts())

                self.ui_assocs.append(assoc)
            except Exception as ex:
                log.warn("Cannot create association for subject=%s pred=%s object=%s: %s" % (sub_refid, pred, obj_refid, ex))

    def publish_device_failed_command_event(self, sub_resource_id, cmd, err_msg):
        """
        PlatformAgent calls this method to publish a DeviceStatusEvent
        indicating that the given child failed to complete the given command.

        @param sub_resource_id   resource id of child (included in values)
        @param cmd               command (included in description)
        @param err_msg           error message (included in description)
        """

        values = [sub_resource_id]
        description = "Child device %r failed to complete command from platform %r (%r)" % \
                      (sub_resource_id, self.resource_id, self._platform_id)
        description += ": cmd=%r; err_msg=%r" % (str(cmd), err_msg)
        evt = dict(event_type='DeviceStatusEvent',
                   sub_type="device_failed_command",
                   origin_type="PlatformDevice",
                   origin=self.resource_id,
                   values=values,
                   description=description)
        try:
            log.debug('%r: publish_device_failed_command_event for %r: %s',
                      self._platform_id, sub_resource_id, evt)

            self._event_publisher.publish_event(**evt)

        except Exception:
            log.exception('%r: platform agent could not publish event: %s',
                          self._platform_id, evt)

    def insert_values(self, coverage, rdt, stream_id):
        elements = len(rdt)

        start_index = coverage.num_timesteps - elements

        for k,v in rdt.iteritems():
            if isinstance(v, SparseConstantValue):
                continue
            slice_ = slice(start_index, None)
            try:
                coverage.set_parameter_values(param_name=k, tdoa=slice_, value=v)
            except IOError as e:
                log.error("Couldn't insert values for coverage: %s",
                          coverage.persistence_dir, exc_info=True)
                try:
                    coverage.close()
                finally:
                    self._bad_coverages[stream_id] = 1
                    raise CorruptionError(e.message)
            except IndexError as e:
                log.error("Value set: %s", v[:])
                data_products, _ = self.container.resource_registry.find_subjects(object=stream_id, predicate=PRED.hasStream, subject_type=RT.DataProduct)
                for data_product in data_products:
                    log.exception("Index exception with %s, trying to insert %s into coverage with shape %s", 
                                  data_product.name,
                                  k,
                                  v.shape)

    
        if 'ingestion_timestamp' in coverage.list_parameters():
            t_now = time.time()
            ntp_time = TimeUtils.ts_to_units(coverage.get_parameter_context('ingestion_timestamp').uom, t_now)
            coverage.set_parameter_values(param_name='ingestion_timestamp', tdoa=slice_, value=ntp_time)

 def kill_mission(self):
     try:
         self.mission_scheduler.kill_mission()
         return None
     except Exception as ex:
         log.exception('[mm] kill_mission')
         return ex

 def _call_request_callback(self, action, req_info):
     if not self.request_callback:
         return
     try:
         self.request_callback(action, req_info)
     except Exception:
         log.exception("Error calling request callback")

    def _get_computed_events(self, events, add_usernames=True, include_events=False):
        """
        Get events for use in extended resource computed attribute
        @retval ComputedListValue with value list of 4-tuple with Event objects
        """
        events = events or []

        ret = IonObject(OT.ComputedEventListValue)
        ret.value = events
        ret.computed_list = [get_event_computed_attributes(event, include_event=include_events) for event in events]
        ret.status = ComputedValueAvailability.PROVIDED

        if add_usernames:
            try:
                actor_ids = {evt.actor_id for evt in events if evt.actor_id}
                log.debug("Looking up UserInfo for actors: %s" % actor_ids)
                if actor_ids:
                    userinfo_list, assoc_list = self.clients.resource_registry.find_objects_mult(actor_ids,
                                                                                                 predicate=PRED.hasInfo,
                                                                                                 id_only=False)
                    actor_map = {assoc.s: uinfo for uinfo, assoc in zip(userinfo_list, assoc_list)}

                    for evt, evt_cmp in zip(events, ret.computed_list):
                        ui = actor_map.get(evt.actor_id, None)
                        if ui:
                            evt_cmp["event_summary"] += " [%s %s]" % (ui.contact.individual_names_given, ui.contact.individual_name_family)

            except Exception as ex:
                log.exception("Cannot find user names for event actor_ids")

        return ret

    def _publish_stream_buffer(self, stream_name):
        """
        overloaded so that data particles are published not granules
        """
        try:
            buf_len = len(self._stream_buffers[stream_name])
            if buf_len == 0:
                return

            publisher = self._publishers[stream_name]

            for x in range(buf_len):
                particle = self._stream_buffers[stream_name].pop()
                publisher.publish(particle)

                log.info("Outgoing particle: %s", particle)

                log.info(
                    "Instrument agent %s published data particle on stream %s.", self._agent._proc_name, stream_name
                )
                log.info(
                    "Connection id: %s, connection index: %i.",
                    self._connection_ID.hex,
                    self._connection_index[stream_name],
                )
        except:
            log.exception(
                "Instrument agent %s could not publish data on stream %s.", self._agent._proc_name, stream_name
            )

    def get_version_info(self, pack=None):
        import pkg_resources
        pkg_list = ["scioncc"]

        packs = self.config.get_safe(CFG_PREFIX + ".version_packages")
        if packs:
            pkg_list.extend(packs.split(","))

        version = {}
        for package in pkg_list:
            try:
                if pack == "all":
                    pack_deps = pkg_resources.require(package)
                    version.update({p.project_name: p.version for p in pack_deps})
                else:
                    version[package] = pkg_resources.require(package)[0].version
                # @TODO git versions for current?
            except pkg_resources.DistributionNotFound:
                pass

        try:
            dir_client = DirectoryServiceProcessClient(process=self.process)
            sys_attrs = dir_client.lookup("/System")
            if sys_attrs and isinstance(sys_attrs, dict):
                version.update({k: v for (k, v) in sys_attrs.iteritems() if "version" in k.lower()})
        except Exception as ex:
            log.exception("Could not determine system directory attributes")

        if pack and pack != "all":
            version = {k: v for (k, v) in version.iteritems() if k == pack}

        return self.gateway_json_response(version)

    def dead_man_timeout(self, stream_id, callback, *args, **kwargs):
        done = False
        timeout = 2
        start = time.time()
        while not done:
            try:
                callback(*args, **kwargs)
                done = True
            except:
                log.exception("An issue with coverage, retrying after a bit")
                if (time.time() - start) > 3600:  # After an hour just give up
                    dataset_id = self.get_dataset(stream_id)
                    log.error(
                        "We're giving up, the coverage needs to be inspected %s",
                        DatasetManagementService._get_coverage_path(dataset_id),
                    )
                    raise

                if stream_id in self._coverages:
                    log.info("Popping coverage for stream %s", stream_id)
                    self._coverages.pop(stream_id)

                gevent.sleep(timeout)
                if timeout > (60 * 5):
                    timeout = 60 * 5
                else:
                    timeout *= 2

    def on_start(self):
        super(IngestionWorker,self).on_start()
        #----------------------------------------------
        # Start up couch
        #----------------------------------------------


        self.couch_config = self.CFG.get('couch_storage')
        self.hdf_storage = self.CFG.get('hdf_storage')

        self.number_of_workers = self.CFG.get('number_of_workers')
        self.description = self.CFG.get('description')

        self.ingest_config_id = self.CFG.get('configuration_id')

        self.datastore_name = self.couch_config.get('datastore_name',None) or 'dm_datastore'
        try:
            self.datastore_profile = getattr(DataStore.DS_PROFILE, self.couch_config.get('datastore_profile','SCIDATA'))
        except AttributeError:
            log.exception('Invalid datastore profile passed to ingestion worker. Defaulting to SCIDATA')

            self.datastore_profile = DataStore.DS_PROFILE.SCIDATA
        log.debug('datastore_profile %s' % self.datastore_profile)
        self.db = self.container.datastore_manager.get_datastore(ds_name=self.datastore_name, profile = self.datastore_profile, config = self.CFG)

        self.resource_reg_client = ResourceRegistryServiceClient(node = self.container.node)

        self.dataset_configs = {}
        # update the policy
        def receive_dataset_config_event(event_msg, headers):
            log.info('Updating dataset config in ingestion worker: %s', event_msg)

            if event_msg.type != DatasetIngestionTypeEnum.DATASETINGESTIONBYSTREAM:
                raise IngestionWorkerException('Received invalid type in dataset config event.')

            stream_id = event_msg.configuration.stream_id

            if event_msg.deleted:
                try:
                    del self.dataset_configs[stream_id]
                except KeyError:
                    log.info('Tried to remove dataset config that does not exist!')
            else:
                self.dataset_configs[stream_id] = event_msg

            # Hook to override just before processing is complete
            self.dataset_configs_event_test_hook(event_msg, headers)


        #Start the event subscriber - really - what a mess!
        self.event_subscriber = EventSubscriber(
            event_type="DatasetIngestionConfigurationEvent",
            origin=self.ingest_config_id,
            callback=receive_dataset_config_event
            )

        self.gl = spawn(self.event_subscriber.listen)
        self.event_subscriber._ready_event.wait(timeout=5)

        log.info(str(self.db))

    def on_restart(self, process, config, **kwargs):
        self.process = process
        inst_objs, _ = process.container.resource_registry.find_resources(restype=RT.Instrument, id_only=False)
        active_agents = []
        for inst in inst_objs:
            if len(inst.agent_state) >= 1:
                active_agents.append(inst._id)

        if not active_agents:
            return

        log.info("Restarting %s agents: %s", len(active_agents), active_agents)

        svc_client = ScionManagementProcessClient(process=process)
        for inst_id in active_agents:
            try:
                svc_client.start_agent(inst_id)
            except Exception as ex:
                log.exception("Cannot restart agent for %s" % inst_id)
                if "Agent already active" in ex.message:
                    try:
                        svc_client.stop_agent(inst_id)
                    except Exception:
                        pass
                    try:
                        svc_client.start_agent(inst_id)
                    except Exception:
                        log.warn("Agent stop/start for %s unsuccessful" % inst_id)