bloodhound-1.0.0.0: tests/Test/AnomalyDetectionSpec.hs
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeApplications #-}
module Test.AnomalyDetectionSpec (spec) where
import Control.Concurrent (threadDelay)
import Control.Monad (void)
import Control.Monad.Catch (bracket_)
import Control.Monad.IO.Class (liftIO)
import Data.Aeson (Value (..), decode, encode, object, (.=))
import Data.ByteString.Lazy.Char8 qualified as LBS
import Data.Fixed (Pico)
import Data.List qualified as L
import Data.Maybe (isJust)
import Data.Text qualified as T
import Data.Time.Calendar (toGregorian)
import Data.Time.Clock (UTCTime (..))
import Data.Time.Clock.POSIX (getPOSIXTime, posixSecondsToUTCTime)
import Data.Time.LocalTime (TimeOfDay (..), timeToTimeOfDay)
import Database.Bloodhound.OpenSearch1.Client qualified as OS1Client
import Database.Bloodhound.OpenSearch1.Requests qualified as OS1Requests
import Database.Bloodhound.OpenSearch1.Types qualified as OS1Types
import Database.Bloodhound.OpenSearch2.Client qualified as OS2Client
import Database.Bloodhound.OpenSearch2.Requests qualified as OS2Requests
import Database.Bloodhound.OpenSearch2.Types qualified as OS2Types
import Database.Bloodhound.OpenSearch3.Client qualified as OS3Client
import Database.Bloodhound.OpenSearch3.Requests qualified as OS3Requests
import Database.Bloodhound.OpenSearch3.Types qualified as OS3Types
import TestsUtils.Common
( os1OnlyIT,
os2OnlyIT,
os3OnlyIT,
withTestEnv,
)
import TestsUtils.Import
import Prelude
-- ---------------------------------------------------------------------------
-- Sample fixtures, drawn verbatim from the OS Anomaly Detection plugin docs
-- (<https://docs.opensearch.org/latest/observing-your-data/ad/api/>).
-- ---------------------------------------------------------------------------
-- | A bare-bones single-entity detector body — exactly the shape the docs
-- show for @POST /_plugins/_anomaly_detection/detectors@ (a name, a
-- @time_field@, the target indices, one feature aggregation, and a
-- detection interval). Used to build both the request 'OS1Types.Detector'
-- and the expected inner @anomaly_detector@ of a decoded response.
sampleSingleEntityDetectorJson :: LBS.ByteString
sampleSingleEntityDetectorJson =
"{\
\ \"name\": \"test-detector\",\
\ \"description\": \"Test detector\",\
\ \"time_field\": \"timestamp\",\
\ \"indices\": [\"anomaly-test\"],\
\ \"feature_attributes\": [\
\ {\
\ \"feature_name\": \"value\",\
\ \"feature_enabled\": true,\
\ \"aggregation_query\": {\
\ \"value\": { \"avg\": { \"field\": \"value\" } }\
\ }\
\ }\
\ ],\
\ \"detection_interval\": {\
\ \"period\": { \"interval\": 1, \"unit\": \"Minutes\" }\
\ },\
\ \"window_delay\": {\
\ \"period\": { \"interval\": 1, \"unit\": \"Minutes\" }\
\ }\
\}"
-- | The create-response envelope returned by the plugin. Captured live
-- against OS 3.7.0 with an OS-generated @rules@ array (a default
-- @IGNORE_ANOMALY@ rule the docs do not enumerate). Carries every
-- server-injected bookkeeping field we model.
sampleCreateResponseJson :: LBS.ByteString
sampleCreateResponseJson =
"{\
\ \"_id\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"_version\": 1,\
\ \"_seq_no\": 0,\
\ \"_primary_term\": 1,\
\ \"anomaly_detector\": {\
\ \"name\": \"test-detector\",\
\ \"description\": \"Test detector\",\
\ \"time_field\": \"timestamp\",\
\ \"indices\": [\"anomaly-test\"],\
\ \"filter_query\": { \"match_all\": { \"boost\": 1.0 } },\
\ \"window_delay\": {\
\ \"period\": { \"interval\": 1, \"unit\": \"Minutes\" }\
\ },\
\ \"shingle_size\": 8,\
\ \"schema_version\": 0,\
\ \"feature_attributes\": [\
\ {\
\ \"feature_id\": \"U0HKTXwBwf_U8gjUXY2m\",\
\ \"feature_name\": \"value\",\
\ \"feature_enabled\": true,\
\ \"aggregation_query\": {\
\ \"value\": { \"avg\": { \"field\": \"value\" } }\
\ }\
\ }\
\ ],\
\ \"recency_emphasis\": 2560,\
\ \"history\": 40,\
\ \"last_update_time\": 1633392680364,\
\ \"detection_interval\": {\
\ \"period\": { \"interval\": 1, \"unit\": \"Minutes\" }\
\ },\
\ \"detector_type\": \"SINGLE_ENTITY\",\
\ \"rules\": [\
\ {\
\ \"action\": \"IGNORE_ANOMALY\",\
\ \"conditions\": [\
\ {\
\ \"feature_name\": \"value\",\
\ \"threshold_type\": \"ACTUAL_OVER_EXPECTED_RATIO\",\
\ \"operator\": \"LTE\",\
\ \"value\": 0.2\
\ }\
\ ]\
\ }\
\ ]\
\ }\
\}"
-- | A high-cardinality / multi-entity detector body. Differs from the
-- single-entity sample only by adding @category_field@; the server
-- derives @detector_type = "MULTI_ENTITY"@ from its presence.
sampleMultiEntityDetectorJson :: LBS.ByteString
sampleMultiEntityDetectorJson =
"{\
\ \"name\": \"test-detector-multi\",\
\ \"description\": \"Multi-entity\",\
\ \"time_field\": \"timestamp\",\
\ \"indices\": [\"anomaly-test\"],\
\ \"category_field\": [\"ip\"],\
\ \"feature_attributes\": [\
\ {\
\ \"feature_name\": \"value\",\
\ \"feature_enabled\": true,\
\ \"aggregation_query\": {\
\ \"value\": { \"avg\": { \"field\": \"value\" } }\
\ }\
\ }\
\ ],\
\ \"detection_interval\": {\
\ \"period\": { \"interval\": 1, \"unit\": \"Minutes\" }\
\ }\
\}"
-- | The @_start@ / @_stop@ response body — the small
-- write-acknowledgment envelope returned by the detector job lifecycle
-- endpoints (@_id@, @_version@, @_seq_no@, @_primary_term@). Drawn
-- verbatim from the Start Detector Job example response in the OS AD
-- API docs.
sampleDetectorJobAcknowledgmentJson :: LBS.ByteString
sampleDetectorJobAcknowledgmentJson =
"{\
\ \"_id\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"_version\": 3,\
\ \"_seq_no\": 6,\
\ \"_primary_term\": 1\
\}"
-- | A 'Value'-typed feature aggregation body, used by both the
-- 'OS1Types.Detector' request sample and the expected
-- 'OS1Types.DetectorResponse' inner object.
sampleAggregationQuery :: Value
sampleAggregationQuery =
object
[ "value"
.= object ["avg" .= object ["field" .= String "value"]]
]
-- | Haskell-side single-entity 'OS1Types.Detector' mirror of
-- 'sampleSingleEntityDetectorJson'. Used for round-trip equality
-- assertions.
os1SampleDetector :: OS1Types.Detector
os1SampleDetector =
OS1Types.Detector
{ OS1Types.detectorName = "test-detector",
OS1Types.detectorDescription = Just "Test detector",
OS1Types.detectorTimeField = "timestamp",
OS1Types.detectorIndices = ["anomaly-test"],
OS1Types.detectorFeatureAttributes =
[ OS1Types.FeatureAttribute
{ OS1Types.featureAttributeId = Nothing,
OS1Types.featureAttributeName = "value",
OS1Types.featureAttributeEnabled = True,
OS1Types.featureAttributeAggregationQuery = sampleAggregationQuery
}
],
OS1Types.detectorFilterQuery = Nothing,
OS1Types.detectorDetectionInterval =
OS1Types.WindowPeriod
{ OS1Types.windowPeriodPeriod =
OS1Types.Period
{ OS1Types.periodInterval = 1,
OS1Types.periodUnit = OS1Types.PeriodUnitMinutes
}
},
OS1Types.detectorWindowDelay =
Just
( OS1Types.WindowPeriod
{ OS1Types.windowPeriodPeriod =
OS1Types.Period
{ OS1Types.periodInterval = 1,
OS1Types.periodUnit = OS1Types.PeriodUnitMinutes
}
}
),
OS1Types.detectorCategoryField = Nothing,
OS1Types.detectorResultIndex = Nothing
}
spec :: Spec
spec = do
-- =======================================================================
-- DetectorId newtype
-- =======================================================================
describe "AnomalyDetection DetectorId JSON" $ do
it "encodes DetectorId as a bare JSON string" $ do
encode (OS1Types.DetectorId "abc-123")
`shouldBe` "\"abc-123\""
it "decodes a bare JSON string into DetectorId" $ do
decode "\"abc-123\"" `shouldBe` Just (OS1Types.DetectorId "abc-123")
it "round-trips DetectorId through encode -> decode" $ do
let original = OS1Types.DetectorId "round-trip-id"
(decode . encode) original `shouldBe` Just original
-- =======================================================================
-- PeriodUnit enum
-- =======================================================================
describe "AnomalyDetection PeriodUnit JSON" $ do
let cases =
[ (OS1Types.PeriodUnitMinutes, "Minutes"),
(OS1Types.PeriodUnitHours, "Hours"),
(OS1Types.PeriodUnitSeconds, "Seconds"),
(OS1Types.PeriodUnitDays, "Days"),
(OS1Types.PeriodUnitWeeks, "Weeks"),
(OS1Types.PeriodUnitMonths, "Months")
]
forM_ cases $ \(u, wire) -> do
it ("encodes " <> show wire <> " to the documented wire string") $ do
encode u `shouldBe` "\"" <> LBS.fromStrict wire <> "\""
it ("decodes the documented wire string " <> show wire) $ do
(decode ("\"" <> LBS.fromStrict wire <> "\"") :: Maybe OS1Types.PeriodUnit)
`shouldBe` Just u
it "decodes an unknown unit into PeriodUnitCustom" $ do
(decode "\"Decades\"" :: Maybe OS1Types.PeriodUnit)
`shouldBe` Just (OS1Types.PeriodUnitCustom "Decades")
it "round-trips PeriodUnit through ToJSON/FromJSON (known)" $ do
let u = OS1Types.PeriodUnitMinutes
(decode . encode) u `shouldBe` Just u
-- =======================================================================
-- DetectorType enum
-- =======================================================================
describe "AnomalyDetection DetectorType JSON" $ do
let cases =
[ (OS1Types.DetectorTypeSingleEntity, "SINGLE_ENTITY"),
(OS1Types.DetectorTypeMultiEntity, "MULTI_ENTITY")
]
forM_ cases $ \(t, wire) -> do
it ("encodes " <> show wire) $ do
encode t `shouldBe` "\"" <> LBS.fromStrict wire <> "\""
it ("decodes " <> show wire) $ do
(decode ("\"" <> LBS.fromStrict wire <> "\"") :: Maybe OS1Types.DetectorType)
`shouldBe` Just t
it "decodes an unknown detector_type into DetectorTypeCustom" $ do
(decode "\"HYBRID\"" :: Maybe OS1Types.DetectorType)
`shouldBe` Just (OS1Types.DetectorTypeCustom "HYBRID")
-- =======================================================================
-- Detector (request body) JSON
-- =======================================================================
describe "AnomalyDetection Detector JSON" $ do
it "decodes the documented single-entity request body" $ do
let decoded = decode sampleSingleEntityDetectorJson :: Maybe OS1Types.Detector
decoded `shouldBe` Just os1SampleDetector
it "round-trips the single-entity Detector through encode -> decode" $ do
(decode . encode) os1SampleDetector `shouldBe` Just os1SampleDetector
it "decodes a multi-entity request body (carries category_field)" $ do
let decoded = decode sampleMultiEntityDetectorJson :: Maybe OS1Types.Detector
decoded `shouldSatisfy` isJust
let Just d = decoded
OS1Types.detectorCategoryField d `shouldBe` Just ["ip"]
it "encodes a Detector omitting nulls (no description, no filter_query)" $ do
let minimal =
os1SampleDetector
{ OS1Types.detectorDescription = Nothing,
OS1Types.detectorWindowDelay = Nothing
}
let encoded = encode minimal
encodedStr = LBS.unpack encoded
-- @description@ and @window_delay@ should NOT appear in the wire form.
encodedStr `shouldSatisfy` not . L.isInfixOf "\"description\""
encodedStr `shouldSatisfy` not . L.isInfixOf "\"window_delay\""
-- @detection_interval@ always appears (required field).
encodedStr `shouldSatisfy` L.isInfixOf "\"detection_interval\""
-- =======================================================================
-- CreateDetectorResponse JSON (envelope + inner DetectorResponse)
-- =======================================================================
describe "AnomalyDetection CreateDetectorResponse JSON" $ do
it "decodes the documented single-entity response envelope (OS1)" $ do
let decoded = decode sampleCreateResponseJson :: Maybe OS1Types.CreateDetectorResponse
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.createDetectorResponseId r `shouldBe` "VEHKTXwBwf_U8gjUXY2s"
OS1Types.createDetectorResponseSeqNo r `shouldBe` 0
OS1Types.createDetectorResponsePrimaryTerm r `shouldBe` 1
let inner = OS1Types.createDetectorResponseDetector r
OS1Types.detectorResponseName inner `shouldBe` "test-detector"
OS1Types.detectorResponseShingleSize inner `shouldBe` Just 8
OS1Types.detectorResponseSchemaVersion inner `shouldBe` Just 0
OS1Types.detectorResponseDetectorType inner
`shouldBe` Just OS1Types.DetectorTypeSingleEntity
OS1Types.detectorResponseRules inner `shouldSatisfy` isJust
it "decodes the same response envelope identically across OS1/OS2/OS3" $ do
let r1 = decode sampleCreateResponseJson :: Maybe OS1Types.CreateDetectorResponse
r2 = decode sampleCreateResponseJson :: Maybe OS2Types.CreateDetectorResponse
r3 = decode sampleCreateResponseJson :: Maybe OS3Types.CreateDetectorResponse
r1 `shouldSatisfy` isJust
r2 `shouldSatisfy` isJust
r3 `shouldSatisfy` isJust
it "round-trips the response envelope through encode -> decode" $ do
let decoded = decode sampleCreateResponseJson :: Maybe OS1Types.CreateDetectorResponse
Just r <- pure decoded
(decode . encode) r `shouldBe` Just r
it "rejects a response missing the anomaly_detector key" $ do
let bad = "{\"_id\":\"x\",\"_version\":1,\"_seq_no\":0,\"_primary_term\":1}"
(decode bad :: Maybe OS1Types.CreateDetectorResponse) `shouldBe` Nothing
-- =======================================================================
-- GetDetectorResponse + GetDetectorParams — the ?job/?task flag surface
-- (bloodhound-ae1). Fixtures drawn from the OS AD Get Detector docs.
-- =======================================================================
describe "AnomalyDetection getDetector request wiring" $ do
let did = OS1Types.DetectorId "abc"
mkParams jobFlag taskFlag =
OS1Types.GetDetectorParams
{ OS1Types.getDetectorParamsJob = jobFlag,
OS1Types.getDetectorParamsTask = taskFlag
}
queriesOf = getRawEndpointQueries . bhRequestEndpoint . OS1Requests.getDetector did
it "default params produce a bare GET with no job/task query" $
queriesOf OS1Types.defaultGetDetectorParams `shouldBe` []
it "?job=true only" $
queriesOf (mkParams True False) `shouldBe` [("job", Just "true")]
it "?task=true only" $
queriesOf (mkParams False True) `shouldBe` [("task", Just "true")]
it "?job=true&task=true together" $
queriesOf (mkParams True True)
`shouldBe` [("job", Just "true"), ("task", Just "true")]
describe "AnomalyDetection GetDetectorResponse JSON" $ do
let baseBody =
"{\
\ \"_id\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"_version\": 1,\
\ \"_primary_term\": 1,\
\ \"_seq_no\": 5,\
\ \"anomaly_detector\": {\
\ \"name\": \"test-detector\",\
\ \"time_field\": \"timestamp\",\
\ \"indices\": [\"anomaly-test\"],\
\ \"detection_interval\": {\
\ \"period\": {\"interval\": 1, \"unit\": \"Minutes\"}\
\ }\
\ }\
\}"
it "decodes the bare response with job/tasks as Nothing" $ do
let decoded = decode baseBody :: Maybe OS1Types.GetDetectorResponse
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.getDetectorResponseId r `shouldBe` "VEHKTXwBwf_U8gjUXY2s"
OS1Types.getDetectorResponseJob r `shouldBe` Nothing
OS1Types.getDetectorResponseRealtimeTask r `shouldBe` Nothing
OS1Types.getDetectorResponseHistoricalTask r `shouldBe` Nothing
it "decodes ?job=true response with only the job populated" $ do
let body =
"{\
\ \"_id\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"_version\": 1,\
\ \"_primary_term\": 1,\
\ \"_seq_no\": 5,\
\ \"anomaly_detector\": {\
\ \"name\": \"test-detector\",\
\ \"time_field\": \"timestamp\",\
\ \"indices\": [\"anomaly-test\"],\
\ \"detection_interval\": {\
\ \"period\": {\"interval\": 1, \"unit\": \"Minutes\"}\
\ }\
\ },\
\ \"anomaly_detector_job\": {\
\ \"name\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"enabled\": true,\
\ \"enabled_time\": 1633393656357,\
\ \"lock_duration_seconds\": 60,\
\ \"schedule\": {\
\ \"interval\": {\
\ \"start_time\": 1633393656357,\
\ \"period\": 1,\
\ \"unit\": \"Minutes\"\
\ }\
\ }\
\ }\
\}"
decoded = decode body :: Maybe OS1Types.GetDetectorResponse
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.detectorJobEnabled <$> OS1Types.getDetectorResponseJob r `shouldBe` Just (Just True)
OS1Types.detectorJobLockDurationSeconds <$> OS1Types.getDetectorResponseJob r `shouldBe` Just (Just 60)
-- schedule typed as the documented {interval:{start_time,period,unit}} form
job <- pure (OS1Types.getDetectorResponseJob r)
OS1Types.detectorJobScheduleIntervalPeriod
<$> (OS1Types.detectorJobScheduleInterval <$> (OS1Types.detectorJobSchedule =<< job))
`shouldBe` Just 1
OS1Types.getDetectorResponseRealtimeTask r `shouldBe` Nothing
OS1Types.getDetectorResponseHistoricalTask r `shouldBe` Nothing
it "decodes ?task=true response with both tasks populated and no job" $ do
let body =
"{\
\ \"_id\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"_version\": 1,\
\ \"_primary_term\": 1,\
\ \"_seq_no\": 5,\
\ \"anomaly_detector\": {\
\ \"name\": \"test-detector\",\
\ \"time_field\": \"timestamp\",\
\ \"indices\": [\"anomaly-test\"],\
\ \"detection_interval\": {\
\ \"period\": {\"interval\": 1, \"unit\": \"Minutes\"}\
\ }\
\ },\
\ \"realtime_detection_task\": {\
\ \"task_id\": \"rt-1\",\
\ \"task_type\": \"REALTIME_SINGLE_ENTITY\",\
\ \"state\": \"RUNNING\",\
\ \"task_progress\": 0,\
\ \"is_latest\": true\
\ },\
\ \"historical_analysis_task\": {\
\ \"task_id\": \"hist-1\",\
\ \"task_type\": \"HISTORICAL_SINGLE_ENTITY\",\
\ \"state\": \"RUNNING\",\
\ \"task_progress\": 0.89285713,\
\ \"current_piece\": 1633328940000,\
\ \"detection_date_range\": {\
\ \"start_time\": 1632788951329,\
\ \"end_time\": 1633393751329\
\ },\
\ \"worker_node\": \"node-x\",\
\ \"is_latest\": true\
\ }\
\}"
decoded = decode body :: Maybe OS1Types.GetDetectorResponse
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.getDetectorResponseJob r `shouldBe` Nothing
-- Real-time task: integer task_progress (0) decodes as Scientific.
rt <- pure (OS1Types.getDetectorResponseRealtimeTask r)
OS1Types.detectionTaskTaskId <$> rt `shouldBe` Just "rt-1"
OS1Types.detectionTaskType
<$> rt
`shouldBe` Just (Just OS1Types.DetectionTaskTypeRealtimeSingleEntity)
-- Historical task: fractional task_progress + historical-only fields.
hist <- pure (OS1Types.getDetectorResponseHistoricalTask r)
OS1Types.detectionTaskTaskId <$> hist `shouldBe` Just "hist-1"
OS1Types.detectionTaskCurrentPiece <$> hist `shouldBe` Just (Just 1633328940000)
OS1Types.detectionTaskWorkerNode <$> hist `shouldBe` Just (Just "node-x")
-- detection_date_range is a historical-only sibling of the task.
ddr <- pure (OS1Types.detectionTaskDetectionDateRange =<< hist)
OS1Types.detectionDateRangeStartTime <$> ddr `shouldBe` Just (Just 1632788951329)
OS1Types.detectionDateRangeEndTime <$> ddr `shouldBe` Just (Just 1633393751329)
it "round-trips the bare response through encode -> decode" $ do
let decoded = decode baseBody :: Maybe OS1Types.GetDetectorResponse
Just r <- pure decoded
(decode . encode) r `shouldBe` Just r
-- =======================================================================
-- DetectionTaskType — enum round-trip + forward-compat escape hatch
-- =======================================================================
describe "AnomalyDetection DetectionTaskType JSON" $ do
let cases =
[ (OS1Types.DetectionTaskTypeRealtimeSingleEntity, "REALTIME_SINGLE_ENTITY"),
(OS1Types.DetectionTaskTypeHistoricalSingleEntity, "HISTORICAL_SINGLE_ENTITY"),
(OS1Types.DetectionTaskTypeRealtimeMultiEntity, "REALTIME_MULTI_ENTITY"),
(OS1Types.DetectionTaskTypeHistoricalMultiEntity, "HISTORICAL_MULTI_ENTITY")
]
it "encodes each documented constructor to its wire string" $
mapM_
( \(con, wire) ->
encode con `shouldBe` "\"" <> LBS.pack wire <> "\""
)
cases
it "decodes each documented wire string" $
mapM_
( \(con, wire) ->
(decode (LBS.pack ("\"" <> wire <> "\"")) :: Maybe OS1Types.DetectionTaskType)
`shouldBe` Just con
)
cases
it "round-trips an unknown value through the Custom escape hatch" $ do
let custom = OS1Types.DetectionTaskTypeCustom "REALTIME_NEW_KIND"
(decode . encode) custom `shouldBe` Just custom
-- =======================================================================
-- DetectorJobAcknowledgment — the _start/_stop job lifecycle response
-- =======================================================================
describe "AnomalyDetection DetectorJobAcknowledgment JSON" $ do
it "decodes the documented @_start@ / @_stop@ envelope" $ do
let decoded = decode sampleDetectorJobAcknowledgmentJson :: Maybe OS1Types.DetectorJobAcknowledgment
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.detectorJobAcknowledgmentId r `shouldBe` "VEHKTXwBwf_U8gjUXY2s"
OS1Types.detectorJobAcknowledgmentSeqNo r `shouldBe` 6
OS1Types.detectorJobAcknowledgmentPrimaryTerm r `shouldBe` 1
it "round-trips through ToJSON / FromJSON (covers @_version@)" $ do
let Just ack = decode sampleDetectorJobAcknowledgmentJson :: Maybe OS1Types.DetectorJobAcknowledgment
(decode . encode) ack `shouldBe` Just (ack :: OS1Types.DetectorJobAcknowledgment)
it "decodes the documented @_stop@ response with zeroed bookkeeping fields" $ do
-- @_version@ is @0@ here; this pins down why the field is 'Int'
-- rather than 'DocVersion' (whose minBound is 1).
let stopResp =
"{\
\ \"_id\": \"VEHKTXwBwf_U8gjUXY2s\",\
\ \"_version\": 0,\
\ \"_seq_no\": 0,\
\ \"_primary_term\": 0\
\}"
decoded = decode stopResp :: Maybe OS1Types.DetectorJobAcknowledgment
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.detectorJobAcknowledgmentId r `shouldBe` "VEHKTXwBwf_U8gjUXY2s"
OS1Types.detectorJobAcknowledgmentVersion r `shouldBe` 0
OS1Types.detectorJobAcknowledgmentSeqNo r `shouldBe` 0
OS1Types.detectorJobAcknowledgmentPrimaryTerm r `shouldBe` 0
it "tolerates an envelope missing @_primary_term@ (OS start/stop acks)" $ do
-- OS 1/2/3 only return @_id@ in start/stop job acknowledgments;
-- the @_version@, @_seq_no@, @_primary_term@ fields are absent.
-- The parser defaults them to 0 to tolerate that wire shape.
let partial = "{\"_id\":\"x\",\"_version\":0,\"_seq_no\":0}"
Just r = decode partial :: Maybe OS1Types.DetectorJobAcknowledgment
OS1Types.detectorJobAcknowledgmentId r `shouldBe` "x"
OS1Types.detectorJobAcknowledgmentPrimaryTerm r `shouldBe` 0
-- =======================================================================
-- StartDetectorJobRequest JSON (historical-analysis body for _start)
-- =======================================================================
describe "AnomalyDetection StartDetectorJobRequest JSON" $ do
it "encodes the documented @start_time@ / @end_time@ field names" $ do
let req =
OS1Types.StartDetectorJobRequest
{ OS1Types.startDetectorJobRequestStartTime = 1633048868000,
OS1Types.startDetectorJobRequestEndTime = 1633394468000
}
encoded = LBS.unpack (encode req)
encoded `shouldSatisfy` L.isInfixOf "\"start_time\":1633048868000"
encoded `shouldSatisfy` L.isInfixOf "\"end_time\":1633394468000"
(decode . encode) req `shouldBe` Just req
it "decodes the documented historical @_start@ body" $ do
let body =
"{\
\ \"start_time\": 1633048868000,\
\ \"end_time\": 1633394468000\
\}"
decoded = decode body :: Maybe OS1Types.StartDetectorJobRequest
decoded
`shouldBe` Just
( OS1Types.StartDetectorJobRequest
{ OS1Types.startDetectorJobRequestStartTime = 1633048868000,
OS1Types.startDetectorJobRequestEndTime = 1633394468000
}
)
-- =======================================================================
-- PreviewRequest / PreviewResponse JSON
-- =======================================================================
describe "AnomalyDetection PreviewRequest JSON" $ do
it "round-trips period_start / period_end (no detector)" $ do
let req =
OS1Types.PreviewRequest
{ OS1Types.previewRequestPeriodStart = 1633048868000,
OS1Types.previewRequestPeriodEnd = 1633394468000,
OS1Types.previewRequestDetector = Nothing,
OS1Types.previewRequestDetectorId = Nothing
}
(decode . encode) req `shouldBe` Just req
it "encodes only the period fields when no detector is set (backward compat)" $ do
let req =
OS1Types.PreviewRequest
{ OS1Types.previewRequestPeriodStart = 1,
OS1Types.previewRequestPeriodEnd = 2,
OS1Types.previewRequestDetector = Nothing,
OS1Types.previewRequestDetectorId = Nothing
}
encoded = LBS.unpack (encode req)
encoded `shouldSatisfy` L.isInfixOf "\"period_start\":1"
encoded `shouldSatisfy` L.isInfixOf "\"period_end\":2"
-- The two new optional fields must NOT appear when 'Nothing', so a
-- plain {id}-in-path preview round-trips exactly as before.
encoded `shouldNotSatisfy` L.isInfixOf "\"detector\""
encoded `shouldNotSatisfy` L.isInfixOf "\"detector_id\""
it "encodes an inline detector (unpersisted config preview)" $ do
let det =
OS1Types.Detector
{ OS1Types.detectorName = "test-detector",
OS1Types.detectorDescription = Nothing,
OS1Types.detectorTimeField = "timestamp",
OS1Types.detectorIndices = ["server_log*"],
OS1Types.detectorFeatureAttributes = [],
OS1Types.detectorFilterQuery = Nothing,
OS1Types.detectorDetectionInterval =
OS1Types.WindowPeriod
(OS1Types.Period 1 OS1Types.PeriodUnitMinutes),
OS1Types.detectorWindowDelay = Nothing,
OS1Types.detectorCategoryField = Nothing,
OS1Types.detectorResultIndex = Nothing
}
req =
OS1Types.PreviewRequest
{ OS1Types.previewRequestPeriodStart = 1,
OS1Types.previewRequestPeriodEnd = 2,
OS1Types.previewRequestDetector = Just det,
OS1Types.previewRequestDetectorId = Nothing
}
encoded = LBS.unpack (encode req)
encoded `shouldSatisfy` L.isInfixOf "\"detector\":{"
-- detector_id must remain absent in the inline form.
encoded `shouldNotSatisfy` L.isInfixOf "\"detector_id\""
(decode . encode) req `shouldBe` Just req
it "encodes a body detector_id (no-id-in-path preview)" $ do
let req =
OS1Types.PreviewRequest
{ OS1Types.previewRequestPeriodStart = 1,
OS1Types.previewRequestPeriodEnd = 2,
OS1Types.previewRequestDetector = Nothing,
OS1Types.previewRequestDetectorId = Just (OS1Types.DetectorId "VEHKTXwBwf_U8gjUXY2s")
}
encoded = LBS.unpack (encode req)
encoded `shouldSatisfy` L.isInfixOf "\"detector_id\":\"VEHKTXwBwf_U8gjUXY2s\""
encoded `shouldNotSatisfy` L.isInfixOf "\"detector\":{"
(decode . encode) req `shouldBe` Just req
it "decodes the documented inline-detector preview body" $ do
let body =
"{\
\ \"period_start\": 1633048868000,\
\ \"period_end\": 1633394468000,\
\ \"detector_id\": \"VEHKTXwBwf_U8gjUXY2s\"\
\}"
decoded = decode body :: Maybe OS1Types.PreviewRequest
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.previewRequestDetectorId r `shouldBe` Just (OS1Types.DetectorId "VEHKTXwBwf_U8gjUXY2s")
OS1Types.previewRequestDetector r `shouldBe` Nothing
describe "AnomalyDetection PreviewResponse JSON" $ do
it "decodes a sparse response (empty anomaly_result list)" $ do
let body =
"{\
\ \"anomaly_result\": [],\
\ \"anomaly_detector\": {\
\ \"name\": \"d\",\
\ \"time_field\": \"ts\",\
\ \"detection_interval\": {\
\ \"period\": {\"interval\": 1, \"unit\": \"Minutes\"}\
\ }\
\ }\
\}"
decoded = decode body :: Maybe OS1Types.PreviewResponse
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.previewResponseAnomalyResult r `shouldBe` []
OS1Types.detectorResponseName (OS1Types.previewResponseAnomalyDetector r)
`shouldBe` "d"
-- =======================================================================
-- Lifecycle response/request JSON: delete / job / validate / search /
-- top-anomalies. Fixtures drawn from the OS AD API reference.
-- =======================================================================
describe "AnomalyDetection DeleteDetectorResponse JSON" $ do
it "decodes the bare ES delete-document envelope (NOT acknowledged)" $ do
let body =
"{\
\ \"_index\": \".opensearch-anomaly-detectors\",\
\ \"_id\": \"70TxTXwBwf_U8gjUXY2s\",\
\ \"_version\": 2,\
\ \"result\": \"deleted\",\
\ \"forced_refresh\": true,\
\ \"_shards\": {\"total\": 2, \"successful\": 2, \"skipped\": 0, \"failed\": 0},\
\ \"_seq_no\": 9,\
\ \"_primary_term\": 1\
\}"
decoded = decode body :: Maybe OS1Types.DeleteDetectorResponse
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.deleteDetectorResponseIndex r `shouldBe` ".opensearch-anomaly-detectors"
OS1Types.deleteDetectorResponseId r `shouldBe` "70TxTXwBwf_U8gjUXY2s"
OS1Types.deleteDetectorResponseResult r `shouldBe` "deleted"
OS1Types.deleteDetectorResponseForcedRefresh r `shouldBe` True
OS1Types.anomalyShardsFailed (OS1Types.deleteDetectorResponseShards r) `shouldBe` 0
OS1Types.deleteDetectorResponseSeqNo r `shouldBe` 9
it "round-trips through encode -> decode" $ do
let decoded = decode "{ \"_index\":\"i\", \"_id\":\"x\", \"_version\":1, \"result\":\"deleted\", \"_shards\":{\"total\":1,\"successful\":1,\"skipped\":0,\"failed\":0}, \"_seq_no\":0, \"_primary_term\":1}" :: Maybe OS1Types.DeleteDetectorResponse
Just r <- pure decoded
(decode . encode) r `shouldBe` Just r
-- \| 'DeletedDocuments' is the response type of both @DELETE \/_delete_by_query@
-- and the AD plugin's @DELETE \/_plugins\/_anomaly_detection\/detectors\/results@.
-- We pin the decoder here against the example body in the AD API reference
-- (<https://docs.opensearch.org/latest/observing-your-data/ad/api/#delete-results>),
-- so a future change to the shared 'DeletedDocuments' decoder that breaks
-- the AD contract surfaces in this spec.
describe "AnomalyDetection DeleteDetectorResults response (DeletedDocuments) JSON" $ do
it "decodes the _delete_by_query envelope from the AD docs example" $ do
let body =
"{\
\ \"took\": 48,\
\ \"timed_out\": false,\
\ \"total\": 28,\
\ \"updated\": 0,\
\ \"created\": 0,\
\ \"deleted\": 28,\
\ \"batches\": 1,\
\ \"version_conflicts\": 0,\
\ \"noops\": 0,\
\ \"retries\": {\
\ \"bulk\": 0,\
\ \"search\": 0\
\ },\
\ \"throttled_millis\": 0,\
\ \"requests_per_second\": -1,\
\ \"throttled_until_millis\": 0,\
\ \"failures\": []\
\}"
decoded = decode body :: Maybe OS1Types.DeletedDocuments
decoded `shouldSatisfy` isJust
let Just r = decoded
OS1Types.delDocsTook r `shouldBe` 48
OS1Types.delDocsTimedOut r `shouldBe` False
OS1Types.delDocsTotal r `shouldBe` 28
OS1Types.delDocsDeleted r `shouldBe` 28
OS1Types.delDocsBatches r `shouldBe` 1
OS1Types.delDocsVersionConflicts r `shouldBe` 0
OS1Types.delDocsNoops r `shouldBe` 0
OS1Types.delDocsRetriesBulk (OS1Types.delDocsRetries r) `shouldBe` 0
OS1Types.delDocsRetriesSearch (OS1Types.delDocsRetries r) `shouldBe` 0
OS1Types.delDocsThrottledMillis r `shouldBe` 0
OS1Types.delDocsRequestsPerSecond r `shouldBe` -1.0
OS1Types.delDocsThrottledUntilMillis r `shouldBe` 0
OS1Types.delDocsFailures r `shouldBe` []
-- The @updated@ / @created@ fields in the docs example are ignored by
-- the shared decoder (always 0 for a delete_by_query); the fields the
-- AD contract relies on are the ones asserted above.
describe "AnomalyDetection ValidateDetectorResponse JSON" $ do
it "decodes the empty-object success body" $ do
let decoded = decode "{}" :: Maybe OS1Types.ValidateDetectorResponse
decoded `shouldSatisfy` isJust
it "decodes a detector-issue body losslessly as Value" $ do
let body = "{\"detector\":{\"feature_attributes\":{\"message\":\"bad\"}}}"
decoded = decode body :: Maybe OS1Types.ValidateDetectorResponse
decoded `shouldSatisfy` isJust
describe "AnomalyDetection TopAnomalies JSON" $ do
it "round-trips a TopAnomaliesRequest omitting nulls" $ do
let req =
( OS1Types.defaultTopAnomaliesRequest
123456789000
987654321000
)
{ OS1Types.topAnomaliesRequestSize = Just 3,
OS1Types.topAnomaliesRequestOrder = Just OS1Types.TopAnomaliesOrderSeverity
}
encoded = LBS.unpack (encode req)
encoded `shouldSatisfy` L.isInfixOf "\"start_time_ms\":123456789000"
encoded `shouldSatisfy` L.isInfixOf "\"end_time_ms\":987654321000"
encoded `shouldSatisfy` L.isInfixOf "\"order\":\"severity\""
-- category_field / task_id are Nothing → omitted.
encoded `shouldNotSatisfy` L.isInfixOf "\"task_id\""
(decode . encode) req `shouldBe` Just req
it "decodes the documented top-anomalies response buckets" $ do
let body =
"{\
\ \"buckets\": [\
\ {\"key\": {\"ip\": \"1.2.3.4\"}, \"doc_count\": 10, \"max_anomaly_grade\": 0.8},\
\ {\"key\": {\"ip\": \"5.6.7.8\"}, \"doc_count\": 12, \"max_anomaly_grade\": 0.6}\
\ ]\
\}"
decoded = decode body :: Maybe OS1Types.TopAnomaliesResponse
Just r <- pure decoded
length (OS1Types.topAnomaliesResponseBuckets r) `shouldBe` 2
let first = head (OS1Types.topAnomaliesResponseBuckets r)
OS1Types.topAnomalyBucketDocCount first `shouldBe` 10
OS1Types.topAnomalyBucketMaxAnomalyGrade first `shouldBe` 0.8
describe "AnomalyDetection SearchDetectorsResponse JSON" $ do
it "decodes a standard search envelope with a typed DetectorResponse _source" $ do
let body =
"{\
\ \"took\": 5,\
\ \"timed_out\": false,\
\ \"_shards\": {\"total\": 1, \"successful\": 1, \"skipped\": 0, \"failed\": 0},\
\ \"hits\": {\
\ \"total\": {\"value\": 1, \"relation\": \"eq\"},\
\ \"max_score\": 1.0,\
\ \"hits\": [\
\ {\
\ \"_index\": \".opensearch-anomaly-detectors\",\
\ \"_id\": \"det-1\",\
\ \"_score\": 1.0,\
\ \"_source\": {\
\ \"name\": \"test-detector\",\
\ \"time_field\": \"timestamp\",\
\ \"detection_interval\": {\"period\": {\"interval\": 1, \"unit\": \"Minutes\"}}\
\ }\
\ }\
\ ]\
\ }\
\}"
decoded = decode body :: Maybe OS1Types.SearchDetectorsResponse
Just r <- pure decoded
OS1Types.anomalySearchResponseTook r `shouldBe` 5
OS1Types.anomalySearchTotalValue (OS1Types.anomalySearchResponseTotal r) `shouldBe` 1
length (OS1Types.anomalySearchResponseHits r) `shouldBe` 1
let onlySource = head (OS1Types.anomalySearchResponseSources r)
OS1Types.detectorResponseName onlySource `shouldBe` "test-detector"
-- =======================================================================
-- Endpoint shape: verify the new AD lifecycle endpoints hit the
-- documented method/path across OS1/OS2/OS3 (the three backends are
-- wire-compatible mirrors — see the ISM mirror convention).
-- =======================================================================
describe "AnomalyDetection lifecycle endpoint shape" $ do
let did1 = OS1Types.DetectorId "d1"
did2 = OS2Types.DetectorId "d2"
did3 = OS3Types.DetectorId "d3"
it "deleteDetector DELETEs /_plugins/_anomaly_detection/detectors/{id} with no body" $ do
bhRequestMethod (OS1Requests.deleteDetector did1) `shouldBe` "DELETE"
getRawEndpoint (bhRequestEndpoint (OS1Requests.deleteDetector did1))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "d1"]
bhRequestBody (OS1Requests.deleteDetector did1) `shouldBe` Nothing
it "updateDetector PUTs /detectors/{id} with the encoded body" $ do
bhRequestMethod (OS1Requests.updateDetector did1 os1SampleDetector) `shouldBe` "PUT"
getRawEndpoint (bhRequestEndpoint (OS1Requests.updateDetector did1 os1SampleDetector))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "d1"]
getRawEndpointQueries (bhRequestEndpoint (OS1Requests.updateDetector did1 os1SampleDetector))
`shouldBe` []
it "updateDetectorWith emits if_seq_no / if_primary_term query params" $ do
let req = OS1Requests.updateDetectorWith did1 os1SampleDetector (Just (7, 4))
L.sortOn fst (getRawEndpointQueries (bhRequestEndpoint req))
`shouldBe` [("if_primary_term", Just "4"), ("if_seq_no", Just "7")]
it "startDetector POSTs /detectors/{id}/_start; empty body for real-time" $ do
bhRequestMethod (OS1Requests.startDetector did1 Nothing) `shouldBe` "POST"
getRawEndpoint (bhRequestEndpoint (OS1Requests.startDetector did1 Nothing))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "d1", "_start"]
bhRequestBody (OS1Requests.startDetector did1 Nothing) `shouldBe` Just ""
it "stopDetector POSTs /detectors/{id}/_stop; adds ?historical=true when historical" $ do
getRawEndpoint (bhRequestEndpoint (OS1Requests.stopDetector did1 False))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "d1", "_stop"]
getRawEndpointQueries (bhRequestEndpoint (OS1Requests.stopDetector did1 False)) `shouldBe` []
getRawEndpointQueries (bhRequestEndpoint (OS1Requests.stopDetector did1 True))
`shouldBe` [("historical", Just "true")]
it "validateDetector POSTs /detectors/_validate (detector) or /_validate/model" $ do
getRawEndpoint (bhRequestEndpoint (OS1Requests.validateDetector OS1Types.ValidateDetector os1SampleDetector))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "_validate"]
getRawEndpoint (bhRequestEndpoint (OS1Requests.validateDetector OS1Types.ValidateModel os1SampleDetector))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "_validate", "model"]
it "topAnomalies GETs /detectors/{id}/results/_topAnomalies with ?historical=" $ do
let treq = OS1Types.defaultTopAnomaliesRequest 1 2
bhRequestMethod (OS1Requests.topAnomalies did1 False treq) `shouldBe` "GET"
getRawEndpoint (bhRequestEndpoint (OS1Requests.topAnomalies did1 False treq))
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "d1", "results", "_topAnomalies"]
getRawEndpointQueries (bhRequestEndpoint (OS1Requests.topAnomalies did1 True treq))
`shouldBe` [("historical", Just "true")]
it "deleteDetector: OS1, OS2, OS3 produce the same path and method" $ do
let same = "same-id"
one = OS1Types.DetectorId same
two = OS2Types.DetectorId same
three = OS3Types.DetectorId same
getRawEndpoint (bhRequestEndpoint (OS1Requests.deleteDetector one))
`shouldBe` getRawEndpoint (bhRequestEndpoint (OS2Requests.deleteDetector two))
getRawEndpoint (bhRequestEndpoint (OS2Requests.deleteDetector two))
`shouldBe` getRawEndpoint (bhRequestEndpoint (OS3Requests.deleteDetector three))
bhRequestMethod (OS1Requests.deleteDetector one)
`shouldBe` bhRequestMethod (OS3Requests.deleteDetector three)
it "deleteDetectorResults DELETEs /_plugins/_anomaly_detection/detectors/results with the encoded body" $ do
let q = object ["query" .= object ["match_all" .= object []]]
req = OS1Requests.deleteDetectorResults q
bhRequestMethod req `shouldBe` "DELETE"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "results"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
-- The body is the aeson-encoded query DSL, not defaulted to {}.
decode (maybe "" id (bhRequestBody req))
`shouldBe` Just q
it "deleteDetectorResults: OS1, OS2, OS3 produce the same path, method and body" $ do
let q = object ["query" .= object ["term" .= object ["detector_id" .= object ["value" .= String "dX"]]]]
one = OS1Requests.deleteDetectorResults q
two = OS2Requests.deleteDetectorResults q
three = OS3Requests.deleteDetectorResults q
getRawEndpoint (bhRequestEndpoint one)
`shouldBe` getRawEndpoint (bhRequestEndpoint two)
getRawEndpoint (bhRequestEndpoint two)
`shouldBe` getRawEndpoint (bhRequestEndpoint three)
bhRequestMethod one `shouldBe` bhRequestMethod three
bhRequestBody one `shouldBe` bhRequestBody three
it "startDetector / stopDetector: OS1, OS2, OS3 produce the same path and method" $ do
let one = OS1Types.DetectorId "x"
two = OS2Types.DetectorId "x"
three = OS3Types.DetectorId "x"
getRawEndpoint (bhRequestEndpoint (OS1Requests.startDetector one Nothing))
`shouldBe` getRawEndpoint (bhRequestEndpoint (OS2Requests.startDetector two Nothing))
getRawEndpoint (bhRequestEndpoint (OS2Requests.startDetector two Nothing))
`shouldBe` getRawEndpoint (bhRequestEndpoint (OS3Requests.startDetector three Nothing))
bhRequestMethod (OS1Requests.stopDetector one False)
`shouldBe` bhRequestMethod (OS3Requests.stopDetector three False)
it "previewDetectorInline POSTs /detectors/_preview (no id in path)" $ do
let req = OS1Requests.previewDetectorInline (OS1Types.PreviewRequest 1 2 Nothing Nothing)
bhRequestMethod req `shouldBe` "POST"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "_preview"]
it "profileDetector: Nothing emits a bodyless GET (backward compat)" $ do
let req = OS1Requests.profileDetector did1 [] False Nothing
bhRequestMethod req `shouldBe` "GET"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_anomaly_detection", "detectors", "d1", "_profile"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
bhRequestBody req `shouldBe` Nothing
it "profileDetector: Just entities emits {\"entity\":[...]} as a GET-with-body" $ do
let entities = [OS1Types.Entity {OS1Types.entityName = "host", OS1Types.entityValue = "i-00f"}]
req = OS1Requests.profileDetector did1 [] False (Just entities)
bhRequestMethod req `shouldBe` "GET"
-- No ?_all when allFlag is False; the entity filter is the only extra.
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
-- The body is the documented {"entity":[{"name":...,"value":...}]}.
let Just body = bhRequestBody req
decode body
`shouldBe` Just
( object
[ "entity"
.= [ object ["name" .= ("host" :: T.Text), "value" .= ("i-00f" :: T.Text)]
]
]
)
it "profileDetector: OS1, OS2, OS3 produce the same path, method and entity body" $ do
let one = OS1Requests.profileDetector (OS1Types.DetectorId "d") [] False (Just [OS1Types.Entity "host" "x"])
two = OS2Requests.profileDetector (OS2Types.DetectorId "d") [] False (Just [OS2Types.Entity "host" "x"])
three = OS3Requests.profileDetector (OS3Types.DetectorId "d") [] False (Just [OS3Types.Entity "host" "x"])
getRawEndpoint (bhRequestEndpoint one)
`shouldBe` getRawEndpoint (bhRequestEndpoint two)
getRawEndpoint (bhRequestEndpoint two)
`shouldBe` getRawEndpoint (bhRequestEndpoint three)
bhRequestMethod one `shouldBe` bhRequestMethod three
bhRequestBody one `shouldBe` bhRequestBody three
-- =======================================================================
-- Live integration: create → get → preview → run round-trip, gated per
-- OS major version. The AD plugin ships with OS 1.0+. Each backend has
-- its own Client/Types module (twin types), so we write one test per
-- backend, gated via osNOnlyIT. A timestamp-suffixed name avoids
-- collisions across runs; bracket_ guarantees the index is torn down
-- even on assertion failure.
-- =======================================================================
describe "AnomalyDetection plugin live round-trip" $ do
os1It <- runIO os1OnlyIT
os1It "create -> get -> preview -> start -> stop (OpenSearch 1)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let indexNameText = T.pack ("bloodhound_test_ad_index_os1_" <> suffix)
detectorNameText = T.pack ("bloodhound_test_ad_det_os1_" <> suffix)
let Right indexName = mkIndexName indexNameText
bracket_
(adSetupIndex indexName)
(void $ performBHRequest $ deleteIndex indexName)
$ do
let detector =
OS1Types.Detector
{ OS1Types.detectorName = detectorNameText,
OS1Types.detectorDescription = Just "bloodhound live test",
OS1Types.detectorTimeField = "timestamp",
OS1Types.detectorIndices = [indexNameText],
OS1Types.detectorFeatureAttributes =
[ OS1Types.FeatureAttribute
{ OS1Types.featureAttributeId = Nothing,
OS1Types.featureAttributeName = "value",
OS1Types.featureAttributeEnabled = True,
OS1Types.featureAttributeAggregationQuery =
object ["value" .= object ["avg" .= object ["field" .= String "value"]]]
}
],
OS1Types.detectorFilterQuery = Nothing,
OS1Types.detectorDetectionInterval =
OS1Types.WindowPeriod
{ OS1Types.windowPeriodPeriod =
OS1Types.Period
{ OS1Types.periodInterval = 1,
OS1Types.periodUnit = OS1Types.PeriodUnitMinutes
}
},
OS1Types.detectorWindowDelay = Nothing,
OS1Types.detectorCategoryField = Nothing,
OS1Types.detectorResultIndex = Nothing
}
createResp <- OS1Client.createDetector detector
case createResp of
Left e ->
liftIO $
expectationFailure
("createDetector failed: " <> T.unpack (errorMessage e))
Right r -> do
let detectorId = OS1Types.DetectorId (OS1Types.createDetectorResponseId r)
getResp <- OS1Client.getDetector detectorId OS1Types.defaultGetDetectorParams
liftIO $
case getResp of
Left e ->
expectationFailure
("getDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
nowMs <- liftIO $ round . (* 1000) <$> getPOSIXTime :: BH IO Integer
let previewReq = OS1Types.PreviewRequest (nowMs - 600000) nowMs Nothing Nothing
previewResp <- OS1Client.previewDetector detectorId previewReq
liftIO $
case previewResp of
Left e ->
expectationFailure
("previewDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
startResp <- OS1Client.startDetector detectorId Nothing
liftIO $
case startResp of
Left e ->
expectationFailure
("startDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
stopResp <- stopDetectorWithRetry (OS1Client.stopDetector detectorId False)
liftIO $
case stopResp of
Left e ->
expectationFailure
("stopDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
-- Lifecycle: update (no job running yet) -> start -> stop -> delete.
updateResp <- OS1Client.updateDetector detectorId detector
liftIO $ expectationFailureOnLeft updateResp "updateDetector failed:"
startResp <- OS1Client.startDetector detectorId Nothing
liftIO $ expectationFailureOnLeft startResp "startDetector failed:"
stopResp <- stopDetectorWithRetry (OS1Client.stopDetector detectorId False)
liftIO $ expectationFailureOnLeft stopResp "stopDetector failed:"
delResp <- deleteDetectorWithRetry (OS1Client.deleteDetector detectorId)
liftIO $ expectationFailureOnLeft delResp "deleteDetector failed:"
os2It <- runIO os2OnlyIT
os2It "create -> get -> preview -> start -> stop (OpenSearch 2)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let indexNameText = T.pack ("bloodhound_test_ad_index_os2_" <> suffix)
detectorNameText = T.pack ("bloodhound_test_ad_det_os2_" <> suffix)
let Right indexName = mkIndexName indexNameText
bracket_
(adSetupIndex indexName)
(void $ performBHRequest $ deleteIndex indexName)
$ do
let detector =
OS2Types.Detector
{ OS2Types.detectorName = detectorNameText,
OS2Types.detectorDescription = Just "bloodhound live test",
OS2Types.detectorTimeField = "timestamp",
OS2Types.detectorIndices = [indexNameText],
OS2Types.detectorFeatureAttributes =
[ OS2Types.FeatureAttribute
{ OS2Types.featureAttributeId = Nothing,
OS2Types.featureAttributeName = "value",
OS2Types.featureAttributeEnabled = True,
OS2Types.featureAttributeAggregationQuery =
object ["value" .= object ["avg" .= object ["field" .= String "value"]]]
}
],
OS2Types.detectorFilterQuery = Nothing,
OS2Types.detectorDetectionInterval =
OS2Types.WindowPeriod
{ OS2Types.windowPeriodPeriod =
OS2Types.Period
{ OS2Types.periodInterval = 1,
OS2Types.periodUnit = OS2Types.PeriodUnitMinutes
}
},
OS2Types.detectorWindowDelay = Nothing,
OS2Types.detectorCategoryField = Nothing,
OS2Types.detectorResultIndex = Nothing
}
createResp <- OS2Client.createDetector detector
case createResp of
Left e ->
liftIO $
expectationFailure
("createDetector failed: " <> T.unpack (errorMessage e))
Right r -> do
let detectorId = OS2Types.DetectorId (OS2Types.createDetectorResponseId r)
getResp <- OS2Client.getDetector detectorId OS2Types.defaultGetDetectorParams
liftIO $
case getResp of
Left e ->
expectationFailure
("getDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
nowMs <- liftIO $ round . (* 1000) <$> getPOSIXTime :: BH IO Integer
let previewReq = OS2Types.PreviewRequest (nowMs - 600000) nowMs Nothing Nothing
previewResp <- OS2Client.previewDetector detectorId previewReq
liftIO $
case previewResp of
Left e ->
expectationFailure
("previewDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
startResp <- OS2Client.startDetector detectorId Nothing
liftIO $
case startResp of
Left e ->
expectationFailure
("startDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
stopResp <- stopDetectorWithRetry (OS2Client.stopDetector detectorId False)
liftIO $
case stopResp of
Left e ->
expectationFailure
("stopDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
-- Lifecycle: update (no job running yet) -> start -> stop -> delete.
updateResp <- OS2Client.updateDetector detectorId detector
liftIO $ expectationFailureOnLeft updateResp "updateDetector failed:"
startResp <- OS2Client.startDetector detectorId Nothing
liftIO $ expectationFailureOnLeft startResp "startDetector failed:"
stopResp <- stopDetectorWithRetry (OS2Client.stopDetector detectorId False)
liftIO $ expectationFailureOnLeft stopResp "stopDetector failed:"
delResp <- deleteDetectorWithRetry (OS2Client.deleteDetector detectorId)
liftIO $ expectationFailureOnLeft delResp "deleteDetector failed:"
os3It <- runIO os3OnlyIT
os3It "create -> get -> preview -> start -> stop (OpenSearch 3)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let indexNameText = T.pack ("bloodhound_test_ad_index_os3_" <> suffix)
detectorNameText = T.pack ("bloodhound_test_ad_det_os3_" <> suffix)
let Right indexName = mkIndexName indexNameText
bracket_
(adSetupIndex indexName)
(void $ performBHRequest $ deleteIndex indexName)
$ do
let detector =
OS3Types.Detector
{ OS3Types.detectorName = detectorNameText,
OS3Types.detectorDescription = Just "bloodhound live test",
OS3Types.detectorTimeField = "timestamp",
OS3Types.detectorIndices = [indexNameText],
OS3Types.detectorFeatureAttributes =
[ OS3Types.FeatureAttribute
{ OS3Types.featureAttributeId = Nothing,
OS3Types.featureAttributeName = "value",
OS3Types.featureAttributeEnabled = True,
OS3Types.featureAttributeAggregationQuery =
object ["value" .= object ["avg" .= object ["field" .= String "value"]]]
}
],
OS3Types.detectorFilterQuery = Nothing,
OS3Types.detectorDetectionInterval =
OS3Types.WindowPeriod
{ OS3Types.windowPeriodPeriod =
OS3Types.Period
{ OS3Types.periodInterval = 1,
OS3Types.periodUnit = OS3Types.PeriodUnitMinutes
}
},
OS3Types.detectorWindowDelay = Nothing,
OS3Types.detectorCategoryField = Nothing,
OS3Types.detectorResultIndex = Nothing
}
createResp <- OS3Client.createDetector detector
case createResp of
Left e ->
liftIO $
expectationFailure
("createDetector failed: " <> T.unpack (errorMessage e))
Right r -> do
let detectorId = OS3Types.DetectorId (OS3Types.createDetectorResponseId r)
getResp <- OS3Client.getDetector detectorId OS3Types.defaultGetDetectorParams
liftIO $
case getResp of
Left e ->
expectationFailure
("getDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
nowMs <- liftIO $ round . (* 1000) <$> getPOSIXTime :: BH IO Integer
let previewReq = OS3Types.PreviewRequest (nowMs - 600000) nowMs Nothing Nothing
previewResp <- OS3Client.previewDetector detectorId previewReq
liftIO $
case previewResp of
Left e ->
expectationFailure
("previewDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
startResp <- OS3Client.startDetector detectorId Nothing
liftIO $
case startResp of
Left e ->
expectationFailure
("startDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
stopResp <- stopDetectorWithRetry (OS3Client.stopDetector detectorId False)
liftIO $
case stopResp of
Left e ->
expectationFailure
("stopDetector failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
-- Lifecycle: update (no job running yet) -> start -> stop -> delete.
updateResp <- OS3Client.updateDetector detectorId detector
liftIO $ expectationFailureOnLeft updateResp "updateDetector failed:"
startResp <- OS3Client.startDetector detectorId Nothing
liftIO $ expectationFailureOnLeft startResp "startDetector failed:"
stopResp <- stopDetectorWithRetry (OS3Client.stopDetector detectorId False)
liftIO $ expectationFailureOnLeft stopResp "stopDetector failed:"
delResp <- deleteDetectorWithRetry (OS3Client.deleteDetector detectorId)
liftIO $ expectationFailureOnLeft delResp "deleteDetector failed:"
-- | Create the target index, seed it with one document, and refresh.
-- The AD plugin refuses to create a detector on an empty index
-- (@Can't create anomaly detector as no document is found in the
-- indices: [...]@), and requires @time_field@ to be mapped as a date.
-- We send @timestamp@ as an ISO-8601 string so the cluster auto-maps it
-- to @date@ on first insert — no explicit mapping PUT needed.
adSetupIndex :: IndexName -> BH IO ()
adSetupIndex indexName = do
_ <-
performBHRequest $
createIndex
(IndexSettings (ShardCount 1) (ReplicaCount 0) defaultIndexMappingsLimits)
indexName
nowMs <- liftIO $ round . (* 1000) <$> getPOSIXTime :: BH IO Integer
let ts = nowMs - 60000
-- ES auto-detects ISO-8601 strings as @date@.
iso = T.pack (iso8601FromEpochMs ts)
_ <-
performBHRequest $
indexDocument
indexName
defaultIndexDocumentSettings
(object ["timestamp" .= String iso, "value" .= Number 1])
(DocId "1")
_ <- performBHRequest $ refreshIndex indexName
pure ()
-- | Render epoch-ms as an ISO-8601 UTC string. The cluster's date parser
-- accepts this format out of the box, so the @timestamp@ field auto-maps
-- to @date@ on first insert (no explicit PUT _mapping needed).
iso8601FromEpochMs :: Integer -> String
iso8601FromEpochMs ms =
let secs = ms `div` 1000
-- Build the components manually to avoid pulling in time-formatting
-- deps beyond what the test already uses.
UTCTime day secsOfDay = posixSecondsToUTCTime (fromIntegral secs)
(y, m, d) = toGregorian day
TimeOfDay hh mm ss = timeToTimeOfDay secsOfDay
pad2 n = let s = show n in if length s == 1 then "0" <> s else s
pad4 n = let s = show n in replicate (max 0 (4 - length s)) '0' <> s
in pad4 (fromIntegral y)
<> "-"
<> pad2 m
<> "-"
<> pad2 d
<> "T"
<> pad2 hh
<> ":"
<> pad2 mm
<> ":"
<> pad2 (floor @Pico ss)
<> "Z"
-- | Fail the test (with a prefix and the cluster error message) when a
-- 'ParsedEsResponse' is 'Left'. Used by the AD live lifecycle assertions
-- so each new endpoint gets a uniform non-Left assertion.
expectationFailureOnLeft :: Either EsError a -> String -> IO ()
expectationFailureOnLeft r prefix =
case r of
Left e -> expectationFailure (prefix <> " " <> T.unpack (errorMessage e))
Right _ -> pure ()
-- | Tolerate the Anomaly Detection plugin's start\/stop race:
-- 'startDetector' returns as soon as it acknowledges the start, but the
-- real-time job is registered asynchronously in the
-- @.opendistro-anomaly-detector-jobs@ index, so an immediate
-- 'stopDetector' can transiently receive "Fail to stop detector" from the
-- server. Retry only that specific error, bounded so a genuine failure
-- still surfaces. Mirrors the 'waitFor' idiom in "Test.CCRSpec".
stopDetectorWithRetry ::
(MonadIO m) =>
m (Either EsError a) ->
m (Either EsError a)
stopDetectorWithRetry stop = go (0 :: Int)
where
go n = do
r <- stop
case r of
Left e
| "Fail to stop detector" `T.isInfixOf` errorMessage e && n < 10 ->
liftIO (threadDelay 1000000) >> go (n + 1)
_ -> pure r
-- | Tolerate the Anomaly Detection plugin's stop\/delete race:
-- 'stopDetector' returns as soon as it acknowledges the stop, but the
-- real-time job document is removed from
-- @.opendistro-anomaly-detector-jobs@ asynchronously, so an immediate
-- 'deleteDetector' can transiently receive "Failed to delete all tasks"
-- from the server (observed on OpenSearch 3.7.0). Retry only that
-- specific error, bounded so a genuine failure still surfaces. Mirrors
-- 'stopDetectorWithRetry'.
deleteDetectorWithRetry ::
(MonadIO m) =>
m (Either EsError a) ->
m (Either EsError a)
deleteDetectorWithRetry del = go (0 :: Int)
where
go n = do
r <- del
case r of
Left e
| "Failed to delete all tasks" `T.isInfixOf` errorMessage e && n < 15 ->
liftIO (threadDelay 1000000) >> go (n + 1)
_ -> pure r