Add an aggregator for IPv4 and IPv6 subnets

[salvatore-campagna]

This PR addresses a request for an IP address subnet aggregator (see #57964).

Parameters accepted by the aggregator include:

• prefix_length (integer, required): defines the network size of the subnet mask
• is_ipv6 (boolean, optional, default: false): defines whether the prefix applies to IPv6 (true) or IPv4 (false) IP addresses
• min_doc_count (integer, optional, default: 1): defines the minimum number of documents for a bucket to be returned in the results
• append_prefix_length (boolean, optional, default: false): defines if the prefix length is appended to the IP address key when returning results
• keyed (boolean, optional, default: false): defines whether the result is returned keyed or as an array of buckets

Each bucket returned by the aggregator represents a different subnet. IPv4 subnets also include a netmask field set to the subnet mask value (i.e. "255.255.0.0" for a /16 subnet).

Related to: #57964 and elastic/kibana#68424

[elasticmachine]

Pinging @elastic/es-analytics-geo (Team:Analytics)

[benwtrent]

👍

[salvatore-campagna]

@elasticmachine run elasticsearch-ci/part-1

[salvatore-campagna]

I am aware I need to add documentation about this new aggregation...I am just waiting for some feedback about the interface.

[flash1293]

The API makes a lot of sense to me - this would be a very good fit for how Kibana does things today 👍

[salvatore-campagna]

This is a sample request including an aggregation on an IPv4 field and an aggregation on an IPv6 field:

curl -X GET "localhost:9200/network/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "size": 0,
  "aggs": {
    "/16": {
      "ip_prefix": {
        "field": "ipv4",
        "prefix_len": 16,
        "is_ipv6": false,
        "min_doc_count": 1,
        "append_prefix_len": false,
        "keyed": false
      }
    },
    "/64": {
      "ip_prefix": {
        "field": "ipv6",
        "prefix_len": 64,
        "is_ipv6": true,
        "min_doc_count": 1,
        "append_prefix_len": false,
        "keyed": false
      }
    }
  }
}'

And this is the response:

{
  "took" : 29,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 5,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "/16" : {
      "buckets" : [
        {
          "key" : "192.168.0.0",
          "netmask" : "255.255.0.0",
          "doc_count" : 4,
          "is_ipv6" : false,
          "prefix_len" : 16
        },
        {
          "key" : "192.169.0.0",
          "netmask" : "255.255.0.0",
          "doc_count" : 1,
          "is_ipv6" : false,
          "prefix_len" : 16
        }
      ]
    },
    "/64" : {
      "buckets" : [
        {
          "key" : "2001:db8:85a3::",
          "doc_count" : 1,
          "is_ipv6" : true,
          "prefix_len" : 64
        },
        {
          "key" : "2001:db8:85ab::",
          "doc_count" : 1,
          "is_ipv6" : true,
          "prefix_len" : 64
        },
        {
          "key" : "2001:dc4:11b2:7::",
          "doc_count" : 1,
          "is_ipv6" : true,
          "prefix_len" : 64
        },
        {
          "key" : "2001:dc4:11b3::",
          "doc_count" : 1,
          "is_ipv6" : true,
          "prefix_len" : 64
        },
        {
          "key" : "2004:db3:85f4:10::",
          "doc_count" : 1,
          "is_ipv6" : true,
          "prefix_len" : 64
        }
      ]
    }
  }
}

[salvatore-campagna]

NOTE: I put the code under org.elasticsearch.search.aggregations.bucket.range which might not be ideal since this is not really a range aggregation (like other range aggregations). Initially this was done supporting multiple ranges and I just left the code there. Looking at other packages I don't see a clear candidate for this...

[not-napoleon]

You can create a new package under o.e.s.aggregations.bucket for this agg. That's a pretty standard pattern for us.

One thing we should probably test - what happens if we get mixed IPv4 and IPv6 addresses in the same run? That doesn't seem like something the user would do on purpose, but for example if they're using an index pattern, it's not hard to imagine that the same field might have changed from IPv4 in one index to IPv6 in another. We should do something sensible in that case, even if it's just failing the shard if the stored data doesn't match what the aggregation expected.

[salvatore-campagna]

I am adding a test were I index IPv6 addresses and then try to aggregate with is_ipv6 = false and viceversa, index documents with IPv4 addresses and try to aggregate with is_ipv6 = true. I think here the result is unpredictable (or at least implementation dependent). The issue is that all IP addresses are encoded as 16-bytes arrays. Then the mask is applied "shifting" the actual mask by 12 bytes if the address is IPv4. I have been thinking of a clever way to handle this so that we don't need to know if the address is IPv4 or IPv6 but could not come up with something good.

[salvatore-campagna]

I was wondering if we should, in some way, encode the IP version in the IpFieldMapper together with the actual IP address (without having to distinguish two different types, IPv4 and IPv6, but still keeping just IP as the data type). I am not sure this is doable. I think this is not worth doing.

[salvatore-campagna]

@elasticmachine update branch

[salvatore-campagna]

@elasticmachine update branch

[salvatore-campagna]

@elasticmachine run elasticsearch-ci/bwc

[salvatore-campagna]

@elasticmachine update branch

[salvatore-campagna]

@elasticmachine run elasticsearch-ci/part-2

[not-napoleon]

Two things that I'd still like to fix before merging. Looking good overall!

[not-napoleon]

config.getValuesSource() should never return null. You want to check config.hasValues(). And you shouldn't even need to do that - ValuesSourceAggregatorFactory#createInternal() will check if there are values to aggregate, and call createUnmapped() if there aren't, which in turn can build a no-op aggregator.

[not-napoleon]

I still don't think we should be allocating short lived objects here. I think you can just pass in the BytesRef and mutate it in place.

[elasticsearchmachine]

Hi @salvatore-campagna, I've created a changelog YAML for you.

[salvatore-campagna]

@elasticmachine update branch

[not-napoleon]

LGTM, thanks!