anchor.diff:  
   
   [[example-clientips]]
  -==== Finding suspicious client IPs by using scripted metrics
  -
  -With {transforms}, you can use 
  -{ref}/search-aggregations-metrics-scripted-metric-aggregation.html[scripted 
  -metric aggregations] on your data. These aggregations are flexible and make 
  -it possible to perform very complex processing. Let's use scripted metrics to 
  -identify suspicious client IPs in the web log sample dataset.
  -
  -We transform the data such that the new index contains the sum of bytes and the 
  -number of distinct URLs, agents, incoming requests by location, and geographic 
  -destinations for each client IP. We also use a scripted field to count the 
  -specific types of HTTP responses that each client IP receives. Ultimately, the 
  -example below transforms web log data into an entity centric index where the 
  -entity is `clientip`.
  +==== Finding suspicious client IPs
  +
  +In this example, we use the web log sample dataset to identify suspicious client 
  +IPs. We transform the data such that the new index contains the sum of bytes and 
  +the number of distinct URLs, agents, incoming requests by location, and 
  +geographic destinations for each client IP. We also use filter aggregations to 
  +count the specific types of HTTP responses that each client IP receives. 
  +Ultimately, the example below transforms web log data into an entity centric 
  +index where the entity is `clientip`.
   
   [source,console]
   ----------------------------------
  
positive.code:       "agent_dc": { "cardinality": { "field": "agent.keyword" }},
        "geo.dest_dc": { "cardinality": { "field": "geo.dest" }},
        "responses.total": { "value_count": { "field": "timestamp" }},
        "responses.counts": { <4>
          "scripted_metric": { 
            "init_script": "state.responses = ['error':0L,'success':0L,'other':0L]",
            "map_script": """
              def code = doc['response.keyword'].value;
              if (code.startsWith('5') || code.startsWith('4')) {
                state.responses.error += 1 ;
              } else if(code.startsWith('2')) {
                state.responses.success += 1;
              } else {
                state.responses.other += 1;
              }
              """,
            "combine_script": "state.responses",
            "reduce_script": """
              def counts = ['error': 0L, 'success': 0L, 'other': 0L];
              for (responses in states) {
                counts.error += responses['error'];
                counts.success += responses['success'];
                counts.other += responses['other'];
              }
              return counts;
              """
            }
          },
        "timestamp.min": { "min": { "field": "timestamp" }},
        "timestamp.max": { "max": { "field": "timestamp" }},
  
negative.code: to synchronize the source and destination indices. The worst case
  ingestion delay is 60 seconds.
  <3> The data is grouped by the `clientip` field.
  <4> Filter aggregation that counts the occurrences of successful (`200`) 
  responses in the `response` field. The following two aggregations (`error404` 
  and `error503`) count the error responses by error codes.
  <5> This `bucket_script` calculates the duration of the `clientip` access based
  on the results of the aggregation.
  
  
  After you create the {transform}, you must start it:
  
  [source,console]
  