(
            self, query_string=None, query_filter=None, query_dsl=None,
            indices=None, return_fields=None, scroll=True)
Search OpenSearch.

        Args:
            query_string: Query string.
            query_filter: Dictionary containing filters to apply.
            query_dsl: Dictionary containing OpenSearch DSL query.
            indices: List of indices to query.
            return_fields: List of fields to return.
            scroll: Boolean determining whether we support scrolling searches
                or not. Defaults to True.

        Returns:
            Generator of Event objects.

        Raises:
            ValueError: if neither query_string or query_dsl is provided.
Search OpenSearch.
Search OpenSearch .
def event_stream(
            self, query_string=None, query_filter=None, query_dsl=None,
            indices=None, return_fields=None, scroll=True):
        """Search OpenSearch.

        Args:
            query_string: Query string.
            query_filter: Dictionary containing filters to apply.
            query_dsl: Dictionary containing OpenSearch DSL query.
            indices: List of indices to query.
            return_fields: List of fields to return.
            scroll: Boolean determining whether we support scrolling searches
                or not. Defaults to True.

        Returns:
            Generator of Event objects.

        Raises:
            ValueError: if neither query_string or query_dsl is provided.
        """
        if not (query_string or query_dsl):
            raise ValueError('Both query_string and query_dsl are missing')

        if not query_filter:
            query_filter = {'indices': self.index_name}

        # If not provided we default to the message field as this will always
        # be present.
        if not return_fields:
            return_fields = ['message']

        # Make sure we always return tag, human_readable and emoji attributes.
        return_fields.extend(['tag', 'human_readable', '__ts_emojis'])
        return_fields = list(set(return_fields))

        if not indices:
            indices = [self.index_name]

        # Refresh the index to make sure it is searchable.
        for index in indices:
            try:
                self.datastore.client.indices.refresh(index=index)
            except opensearchpy.NotFoundError:
                logger.error(
                    'Unable to find index: {0:s}, removing from '
                    'result set.'.format(index))
                broken_index = indices.index(index)
                _ = indices.pop(broken_index)
        if not indices:
            raise ValueError(
                'Unable to query for analyzers, discovered no index to query.')

        if self.timeline_id:
            timeline_ids = [self.timeline_id]
        else:
            timeline_ids = None

        # Exponential backoff for the call to OpenSearch. Sometimes the
        # cluster can be a bit overloaded and timeout on requests. We want to
        # retry a few times in order to give the cluster a chance to return
        # results.
        backoff_in_seconds = 3
        retries = 5
        for x in range(0, retries):
            try:
                event_generator = self.datastore.search_stream(
                    query_string=query_string,
                    query_filter=query_filter,
                    query_dsl=query_dsl,
                    indices=indices,
                    return_fields=return_fields,
                    enable_scroll=scroll,
                    timeline_ids=timeline_ids
                )
                for event in event_generator:
                    yield Event(
                        event, self.datastore, sketch=self.sketch,
                        analyzer=self)
                break  # Query was succesful
            except opensearchpy.TransportError as e:
                sleep_seconds = (
                    backoff_in_seconds * 2 ** x + random.uniform(3, 7))
                logger.info(
                    'Attempt: {0:d}/{1:d} sleeping {2:f} for query {3:s}'
                    .format(x + 1, retries, sleep_seconds, query_string))
                time.sleep(sleep_seconds)

                if x == retries-1:
                    logger.error(
                        'Timeout executing search for {0:s}: {1!s}'
                        .format(query_string, e), exc_info=True
                    )
                    raise
def event_stream ( self , query_string = None , query_filter = None , query_dsl = None , indices = None , return_fields = None , scroll = True ) : if not ( query_string or query_dsl ) : raise ValueError ( 'Both query_string and query_dsl are missing' ) if not query_filter : query_filter = { 'indices' : self . index_name } # If not provided we default to the message field as this will always # be present. if not return_fields : return_fields = [ 'message' ] # Make sure we always return tag, human_readable and emoji attributes. return_fields . extend ( [ 'tag' , 'human_readable' , '__ts_emojis' ] ) return_fields = list ( set ( return_fields ) ) if not indices : indices = [ self . index_name ] # Refresh the index to make sure it is searchable. for index in indices : try : self . datastore . client . indices . refresh ( index = index ) except opensearchpy . NotFoundError : logger . error ( 'Unable to find index: {0:s}, removing from ' 'result set.' . format ( index ) ) broken_index = indices . index ( index ) _ = indices . pop ( broken_index ) if not indices : raise ValueError ( 'Unable to query for analyzers, discovered no index to query.' ) if self . timeline_id : timeline_ids = [ self . timeline_id ] else : timeline_ids = None # Exponential backoff for the call to OpenSearch. Sometimes the # cluster can be a bit overloaded and timeout on requests. We want to # retry a few times in order to give the cluster a chance to return # results. backoff_in_seconds = 3 retries = 5 for x in range ( 0 , retries ) : try : event_generator = self . datastore . search_stream ( query_string = query_string , query_filter = query_filter , query_dsl = query_dsl , indices = indices , return_fields = return_fields , enable_scroll = scroll , timeline_ids = timeline_ids ) for event in event_generator : yield Event ( event , self . datastore , sketch = self . sketch , analyzer = self ) break # Query was succesful except opensearchpy . TransportError as e : sleep_seconds = ( backoff_in_seconds * 2 ** x + random . uniform ( 3 , 7 ) ) logger . info ( 'Attempt: {0:d}/{1:d} sleeping {2:f} for query {3:s}' . format ( x + 1 , retries , sleep_seconds , query_string ) ) time . sleep ( sleep_seconds ) if x == retries - 1 : logger . error ( 'Timeout executing search for {0:s}: {1!s}' . format ( query_string , e ) , exc_info = True ) raise