Support Elasticsearch 8

[jertel]

Elasticsearch v8 no longer supports doc_type. There is likely going to be some effort need to update this project to deal with this.

[nsano-rururu]

ES8 support may need to be considered in addition to doc_type.

In ES7.X this include_type_name default has been changed to false, but elastalert explicitly specifies include_type_name = true to use the type, but it will be removed in ES8. Will be.

Also, the specification of dateOptionalTime is abolished and becomes date_optional_time.

elasticsearch>=7.0.0,<8.0.0

    ProcessController:  /home/node/.local/lib/python3.8/site-packages/elasticsearch/connection/base.py:193: ElasticsearchDeprecationWarning: Camel case format name dateOptionalTime is deprecated and will be removed in a future version. Use snake case name date_optional_time instead.
      warnings.warn(message, category=ElasticsearchDeprecationWarning)
    /home/node/.local/lib/python3.8/site-packages/elasticsearch/connection/base.py:193: ElasticsearchDeprecationWarning: [types removal] Using include_type_name in put mapping requests is deprecated. The parameter will be removed in the next major version.
      warnings.warn(message, category=ElasticsearchDeprecationWarning)

---

The following are expected changes that need to be addressed.

___init___.py

add

    def is_atleasteight(self):
        """
        Returns True when the Elasticsearch server version >= 8
        """
        return int(self.es_version.split(".")[0]) >= 8

elastalert/create_index.py

before

   if is_atleastseven(esversion):
        # TODO remove doc_type completely when elasicsearch client allows doc_type=None
        # doc_type is a deprecated feature and will be completely removed in Elasicsearch 8
        es_client.indices.put_mapping(index=ea_index, doc_type='_doc',
                                      body=es_index_mappings['elastalert'], include_type_name=True)
        es_client.indices.put_mapping(index=ea_index + '_status', doc_type='_doc',
                                      body=es_index_mappings['elastalert_status'], include_type_name=True)
        es_client.indices.put_mapping(index=ea_index + '_silence', doc_type='_doc',
                                      body=es_index_mappings['silence'], include_type_name=True)
        es_client.indices.put_mapping(index=ea_index + '_error', doc_type='_doc',
                                      body=es_index_mappings['elastalert_error'], include_type_name=True)
        es_client.indices.put_mapping(index=ea_index + '_past', doc_type='_doc',
                                      body=es_index_mappings['past_elastalert'], include_type_name=True)

after

   if is_atleasteight(esversion):
        es_client.indices.put_mapping(index=ea_index,
                                      body=es_index_mappings['elastalert'])
        es_client.indices.put_mapping(index=ea_index + '_status'
                                      body=es_index_mappings['elastalert_status'])
        es_client.indices.put_mapping(index=ea_index + '_silence',
                                      body=es_index_mappings['silence'])
        es_client.indices.put_mapping(index=ea_index + '_error',
                                      body=es_index_mappings['elastalert_error'])
        es_client.indices.put_mapping(index=ea_index + '_past',
                                      body=es_index_mappings['past_elastalert'])
   elif is_atleastseven(esversion):
        es_client.indices.put_mapping(index=ea_index, doc_type='_doc',
                                      body=es_index_mappings['elastalert'])
        es_client.indices.put_mapping(index=ea_index + '_status', doc_type='_doc',
                                      body=es_index_mappings['elastalert_status'])
        es_client.indices.put_mapping(index=ea_index + '_silence', doc_type='_doc',
                                      body=es_index_mappings['silence'])
        es_client.indices.put_mapping(index=ea_index + '_error', doc_type='_doc',
                                      body=es_index_mappings['elastalert_error'])
        es_client.indices.put_mapping(index=ea_index + '_past', doc_type='_doc',
                                      body=es_index_mappings['past_elastalert'])

elastalert/loaders.py

It is necessary to prevent the following checks from being performed on elasticsearch 8 or later versions

        if rule.get('use_count_query') or rule.get('use_terms_query'):
            if 'doc_type' not in rule:
                raise EAException('doc_type must be specified.')

elastalert/elastalert.py

before

        # Record doc_type for use in get_top_counts
        if 'doc_type' not in rule and len(hits):
          rule['doc_type'] = hits[0]['_type']

after

        # Record doc_type for use in get_top_counts
        if not self.thread_data.current_es.is_atleasteight():
            if 'doc_type' not in rule and len(hits):
                rule['doc_type'] = hits[0]['_type']

before

            if not rule['five']:
                res = self.thread_data.current_es.deprecated_search(
                    index=index,
                    doc_type=rule['doc_type'],
                    body=query,
                    search_type='count',
                    ignore_unavailable=True
                )
            else:
                res = self.thread_data.current_es.deprecated_search(index=index, doc_type=rule['doc_type'],
                                                                    body=query, size=0, ignore_unavailable=True)

after

            if not rule['five']:
                if self.thread_data.current_es.is_atleasteight():
                    res = self.thread_data.current_es.deprecated_search(
                        index=index,
                        body=query,
                        search_type='count',
                        ignore_unavailable=True
                    )
                else:
                    res = self.thread_data.current_es.deprecated_search(
                        index=index,
                        doc_type=rule['doc_type'],
                        body=query,
                        search_type='count',
                        ignore_unavailable=True
                    )
            else:
                res = self.thread_data.current_es.deprecated_search(index=index, doc_type=rule['doc_type'],
                                                                    body=query, size=0, ignore_unavailable=True)

before

            if not rule['five']:
                res = self.thread_data.current_es.deprecated_search(
                    index=index,
                    doc_type=rule.get('doc_type'),
                    body=query,
                    search_type='count',
                    ignore_unavailable=True
                )
            else:
                res = self.thread_data.current_es.deprecated_search(index=index, doc_type=rule.get('doc_type'),
                                                                    body=query, size=0, ignore_unavailable=True)

after

            if not rule['five']:
                if self.thread_data.current_es.is_atleasteight():
                    res = self.thread_data.current_es.deprecated_search(
                        index=index,
                        body=query,
                        search_type='count',
                        ignore_unavailable=True
                    )
                else:
                    res = self.thread_data.current_es.deprecated_search(
                        index=index,
                        doc_type=rule.get('doc_type'),
                        body=query,
                        search_type='count',
                        ignore_unavailable=True
                    )
            else:
                res = self.thread_data.current_es.deprecated_search(index=index, doc_type=rule.get('doc_type'),
                                                                    body=query, size=0, ignore_unavailable=True)

elastalert/test_rule.py

before

res = es_client.count(index=index, doc_type=doc_type, body=count_query, ignore_unavailable=True)

after

if es_client.is_atleasteight():
    res = es_client.count(index=index, body=count_query, ignore_unavailable=True)
else:
    res = es_client.count(index=index, doc_type=doc_type, body=count_query, ignore_unavailable=True)

elastalert/tests/conftest.py

mock_es_client

add

self.is_atleasteight = mock.Mock(return_value=False)

mock_es_sixsix_client

add

self.is_atleasteight = mock.Mock(return_value=False)

elastalert/es_mappings/8/elastalert.json

add file

dateOptionalTime→date_optional_time

{
  "numeric_detection": true,
  "date_detection": false,
  "dynamic_templates": [
    {
      "strings_as_keyword": {
        "mapping": {
          "ignore_above": 1024,
          "type": "keyword"
        },
        "match_mapping_type": "string"
      }
    }
  ],
  "properties": {
    "rule_name": {
      "type": "keyword"
    },
    "@timestamp": {
      "type": "date",
      "format": "date_optional_time"
    },
    "alert_time": {
      "type": "date",
      "format": "date_optional_time"
    },
    "match_time": {
      "type": "date",
      "format": "date_optional_time"
    },
    "match_body": {
      "enabled": "false",
      "type": "object"
    },
    "aggregate_id": {
      "type": "keyword"
    }
  }
}

add Directory

elastalert/es_mappings/8

elastalert/es_mappings/8/elastalert_error.json ・・・add file

dateOptionalTime→date_optional_time

{
  "properties": {
    "data": {
      "type": "object",
      "enabled": "false"
    },
    "@timestamp": {
      "type": "date",
      "format": "date_optional_time"
    }
  }
}

elastalert/es_mappings/8/elastalert_status.json・・・add file

dateOptionalTime→date_optional_time

{
  "properties": {
    "rule_name": {
      "type": "keyword"
    },
    "@timestamp": {
      "type": "date",
      "format": "date_optional_time"
    }
  }
}

elastalert/es_mappings/8/past_elastalert.json・・・add file

dateOptionalTime→date_optional_time

{
  "properties": {
    "rule_name": {
      "type": "keyword"
    },
    "match_body": {
      "type": "object",
      "enabled": "false"
    },
    "@timestamp": {
      "type": "date",
      "format": "date_optional_time"
    },
    "aggregate_id": {
      "type": "keyword"
    }
  }
}

elastalert/es_mappings/8/silence.json・・・add file

dateOptionalTime→date_optional_time

{
  "properties": {
    "rule_name": {
      "type": "keyword"
    },
    "until": {
      "type": "date",
      "format": "date_optional_time"
    },
    "@timestamp": {
      "type": "date",
      "format": "date_optional_time"
    }
  }
}

elastalert/create_index.py

before

es_index_mappings = read_es_index_mappings() if is_atleastsix(esversion) else read_es_index_mappings(5)

after

es_index_mappings = read_es_index_mappings() is_atleasteight(esversion) elif is_atleastsix(esversion) else read_es_index_mappings(5)

tests/create_index_test.py

add test_read_es_8_index_mappings

def test_read_es_8_index_mappings():
    mappings = elastalert.create_index.read_es_index_mappings(8)
    assert len(mappings) == len(es_mappings)
    print((json.dumps(mappings, indent=2)))

[nsano-rururu]

If elasticsearch 8 is released and supported, the docker image will likely need to provide both pip install elasticsearch==7.0.0 for elasticsearch 7 and pip install elasticsearch==8.0.0 for elasticsearch 8.

[nsano-rururu]

This is the exact code block that needs to be changed:
loaders.py
YelpArchive/elastalert#2424 (comment)

        # Check that doc_type is provided if use_count/terms_query
        if rule.get('use_count_query') or rule.get('use_terms_query'):
            if 'doc_type' not in rule:
                raise EAException('doc_type must be specified.')

[ferozsalam]

I noticed the alpha ES 8 image is now available from in the Elasticsearch Docker registry. I'll start working on ES 8 compatibility this weekend, assuming no one else is already looking at this.

[jertel]

Thanks @ferozsalam.

[nsano-rururu]

@ferozsalam

Since elasticsearch-py 8.0.0 of elasticsearch's pytnon client has not been released yet, is it correct to change the code once in the state of elasticsearch-py 7.0.0 and check the operation?
https://github.com/elastic/elasticsearch-py

[ferozsalam]

That's a good point.

I think my plan will be to see if I can get things working with the ES8 alpha and elasticsearch-py 7.0.0 over the weekend. Elastic has already started work on elasticsearch-py 8.0.0, so if 7.0.0 doesn't work I might try using the latest version of the library direct from the repository.

If it all doesn't work and we need to wait for elasticsearch-py 8.0.0, I'll post here and pause for a while.

[nsano-rururu]

Since the latest elasticsearch-py currently has an implementation that does not connect to Amazon Elasticsearch Service, it is necessary to support opensearch-py at the same time as supporting elasticsearch-py 8.
https://aws.amazon.com/jp/blogs/opensource/keeping-clients-of-opensearch-and-elasticsearch-compatible-with-open-source/
https://github.com/opensearch-project/opensearch-py

opensearch-py seems to be able to connect to both elasticsearch and Amazon Elasticsearch Service, but the client version before the connection restrictions were built in was elasticsearch-py 7.13.4. Please note that elastalert2 uses elasticsearch-py7.0.0, so if you change all connections to opensearch-py, some rules will not work.
It's possible to determine the Amazon Elasticsearch Service, but you need to consider what to do if you use opensearch alone.

https://opensearch.org/docs/clients/index/
OpenSearch client compatibility

Python Elasticsearch client 7.13.4

[nsano-rururu]

@ferozsalam

The index kibana-int described in the code of elastalert/rule_from_kibana.py and elastalert/elastalert.py does not exist in Elasticsearch 7 (also Elasticsearch 6?), So there are some errors in the results I analyzed in the past.
#92 (comment)
Please refer to the content of the discussion commented by jertel.
#442 (reply in thread)

[nsano-rururu]

@ferozsalam

Share information about es 8.

Deprecation warnings in 7.15.0 pre-releases
elastic/elasticsearch-py#1698

The body parameter for APIs are deprecated

I didn't expect the API body parameter to be deprecated. I think it has a big impact.

[ferozsalam]

I've spent some time this weekend (much delayed!) getting a development environment setup, and have managed to get ElastAlert running and communicating to Elasticsearch 8.0.0-alpha2 using elasticsearch-py 7.0.0 with a single debug rule running. The good news is that it looks like ES 8-alpha2 still works (at least for ElastAlert) with elasticsearch-py 7.0.0.

There are several places where I need to make further tweaks to handle the removal of doc types, but I don't foresee that proving a major hurdle.

My goal for this week is to get the unit test suite working to see what else needs fixing/changing. I will probably start doing the development work on a separate branch.

Thanks very much for your code samples above @nsano-rururu - they saved me a lot of time.

A question - does anyone know why we have pinned the current elasticsearch-py version at 7.0.0? We're probably going to run into whatever issue is causing that again with the conversion to ES 8, so I wondered if it might be good to also fix that if possible.

[nsano-rururu]

YelpArchive/elastalert#2593 (comment)

Also note: I also added a pin for elasticsearch==7.0.0, because apparently 7.1.0 will NOT work with ES < 6.6 due to it not supported _source_include(s?). 7.0.0 does. Tests won't pass otherwise.

Fix issue caused by 7.x breaking change (_source_include/_source_exclude)
elastic/elasticsearch-py#1019

https://www.elastic.co/guide/en/elasticsearch/reference/7.0/breaking-changes-7.0.html#source-include-exclude-params-removed

Source filtering url parameters _source_include and _source_exclude have been removed
The deprecated in 6.x url parameters are now removed. Use _source_includes and _source_excludes instead.

ES Version revert to 7.0.0
#90