bloodhound-1.0.0.0: tests/Test/RollupsSpec.hs
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeApplications #-}
module Test.RollupsSpec (spec) where
import Control.Monad.Catch (bracket_)
import Data.ByteString.Lazy.Char8 qualified as LBS
import Data.Fixed (Pico)
import Data.List (sort)
import Data.Text qualified as T
import Data.Time.Calendar (toGregorian)
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 Index Rollups plugin docs
-- (<https://docs.opensearch.org/latest/im-plugin/index-rollups/rollup-api/>).
-- ---------------------------------------------------------------------------
-- | A minimal create-request body wrapped in the @{"rollup": ...}@ envelope.
sampleCreateRequestJson :: LBS.ByteString
sampleCreateRequestJson =
"{\
\ \"rollup\": {\
\ \"source_index\": \"nyc-taxi-data\",\
\ \"target_index\": \"rollup-nyc-taxi-data\",\
\ \"schedule\": {\
\ \"interval\": { \"period\": 1, \"unit\": \"Days\" }\
\ },\
\ \"description\": \"Example rollup job\",\
\ \"enabled\": true,\
\ \"page_size\": 200,\
\ \"delay\": 0,\
\ \"continuous\": false,\
\ \"dimensions\": [\
\ { \"date_histogram\": {\
\ \"source_field\": \"tpep_pickup_datetime\",\
\ \"fixed_interval\": \"1h\",\
\ \"timezone\": \"America/Los_Angeles\"\
\ } },\
\ { \"terms\": { \"source_field\": \"PULocationID\" } }\
\ ],\
\ \"metrics\": [\
\ { \"source_field\": \"passenger_count\",\
\ \"metrics\": [ {\"avg\":{}}, {\"sum\":{}}, {\"max\":{}} ] }\
\ ]\
\ }\
\}"
-- | The PUT response envelope (snake_case @_seq_no@\/@_primary_term@).
samplePutResponseJson :: LBS.ByteString
samplePutResponseJson =
"{\
\ \"_id\": \"example_rollup\",\
\ \"_version\": 3,\
\ \"_seq_no\": 1,\
\ \"_primary_term\": 1,\
\ \"rollup\": {\
\ \"rollup_id\": \"example_rollup\",\
\ \"enabled\": true,\
\ \"schedule\": {\
\ \"interval\": {\
\ \"start_time\": 1680159934649,\
\ \"period\": 1,\
\ \"unit\": \"Days\",\
\ \"schedule_delay\": 0\
\ }\
\ },\
\ \"last_updated_time\": 1680159934649,\
\ \"enabled_time\": 1680159934649,\
\ \"schema_version\": 17,\
\ \"source_index\": \"nyc-taxi-data\",\
\ \"target_index\": \"rollup-nyc-taxi-data\",\
\ \"metadata_id\": null,\
\ \"page_size\": 200,\
\ \"delay\": 0,\
\ \"continuous\": false,\
\ \"dimensions\": [\
\ { \"date_histogram\": {\
\ \"fixed_interval\": \"1h\",\
\ \"source_field\": \"tpep_pickup_datetime\",\
\ \"target_field\": \"tpep_pickup_datetime\",\
\ \"timezone\": \"America/Los_Angeles\"\
\ } },\
\ { \"terms\": {\
\ \"source_field\": \"PULocationID\",\
\ \"target_field\": \"PULocationID\"\
\ } }\
\ ],\
\ \"metrics\": [\
\ { \"source_field\": \"passenger_count\",\
\ \"metrics\": [ {\"avg\":{}}, {\"sum\":{}} ] }\
\ ]\
\ }\
\}"
-- | The GET response envelope (camelCase @_seqNo@\/@_primaryTerm@) — the
-- documented quirk that 'RollupResponse' must also decode.
sampleGetResponseJson :: LBS.ByteString
sampleGetResponseJson =
"{\
\ \"_id\": \"example_rollup\",\
\ \"_version\": 3,\
\ \"_seqNo\": 1,\
\ \"_primaryTerm\": 1,\
\ \"rollup\": {\
\ \"rollup_id\": \"example_rollup\",\
\ \"enabled\": true,\
\ \"schedule\": {\
\ \"interval\": { \"period\": 1, \"unit\": \"Days\" }\
\ },\
\ \"source_index\": \"nyc-taxi-data\",\
\ \"target_index\": \"rollup-nyc-taxi-data\",\
\ \"page_size\": 200\
\ }\
\}"
-- | The explain response when the job has not yet executed (both fields null).
sampleExplainNullJson :: LBS.ByteString
sampleExplainNullJson =
"{\
\ \"example_rollup\": {\
\ \"metadata_id\": null,\
\ \"rollup_metadata\": null\
\ }\
\}"
-- | The explain response after the job has executed.
sampleExplainExecutedJson :: LBS.ByteString
sampleExplainExecutedJson =
"{\
\ \"example_rollup\": {\
\ \"metadata_id\": \"GtWGlZwBm3bOohSSvi2r\",\
\ \"rollup_metadata\": {\
\ \"rollup_id\": \"example_rollup\",\
\ \"last_updated_time\": 1772035161995,\
\ \"status\": \"finished\",\
\ \"failure_reason\": null,\
\ \"stats\": {\
\ \"pages_processed\": 2,\
\ \"documents_processed\": 3,\
\ \"rollups_indexed\": 3,\
\ \"index_time_in_millis\": 28,\
\ \"search_time_in_millis\": 46\
\ }\
\ }\
\ }\
\}"
metricAggTable :: [(OS1Types.RollupMetricAgg, LBS.ByteString)]
metricAggTable =
[ (OS1Types.RollupMetricAvg, "{\"avg\":{}}"),
(OS1Types.RollupMetricSum, "{\"sum\":{}}"),
(OS1Types.RollupMetricMax, "{\"max\":{}}"),
(OS1Types.RollupMetricMin, "{\"min\":{}}"),
(OS1Types.RollupMetricValueCount, "{\"value_count\":{}}")
]
statusTable :: [(OS1Types.RollupStatus, T.Text)]
statusTable =
[ (OS1Types.RollupStatusInit, "init"),
(OS1Types.RollupStatusStarted, "started"),
(OS1Types.RollupStatusFinished, "finished"),
(OS1Types.RollupStatusFailed, "failed"),
(OS1Types.RollupStatusStopped, "stopped"),
(OS1Types.RollupStatusRetry, "retry")
]
intervalUnitTable :: [(OS1Types.RollupIntervalUnit, T.Text)]
intervalUnitTable =
[ (OS1Types.RollupIntervalUnitMinutes, "Minutes"),
(OS1Types.RollupIntervalUnitHours, "Hours"),
(OS1Types.RollupIntervalUnitDays, "Days"),
(OS1Types.RollupIntervalUnitWeeks, "Weeks"),
(OS1Types.RollupIntervalUnitMonths, "Months")
]
-- | A minimal OS1 'Rollup' for the pure endpoint-shape tests. Only the
-- required fields are populated; everything else is 'Nothing'/empty.
minimalOS1Rollup :: OS1Types.Rollup
minimalOS1Rollup =
OS1Types.Rollup
{ OS1Types.rollupSourceIndex = "src",
OS1Types.rollupTargetIndex = "tgt",
OS1Types.rollupTargetIndexSettings = Nothing,
OS1Types.rollupSchedule =
OS1Types.RollupSchedule $
OS1Types.RollupInterval
{ OS1Types.rollupIntervalPeriod = Just 1,
OS1Types.rollupIntervalUnit = Just OS1Types.RollupIntervalUnitMinutes,
OS1Types.rollupIntervalStartTime = Nothing,
OS1Types.rollupIntervalScheduleDelay = Nothing,
OS1Types.rollupIntervalCron = Nothing
},
OS1Types.rollupDescription = Nothing,
OS1Types.rollupEnabled = Nothing,
OS1Types.rollupContinuous = Nothing,
OS1Types.rollupPageSize = 100,
OS1Types.rollupDelay = Nothing,
OS1Types.rollupRoles = Nothing,
OS1Types.rollupRoutingField = Nothing,
OS1Types.rollupDimensions = [],
OS1Types.rollupMetrics = Nothing,
OS1Types.rollupErrorNotification = Nothing
}
spec :: Spec
spec = describe "OpenSearch Index Rollups API" $ do
-- ========================================================================
-- Pure JSON round-trip / wire-shape tests (no live cluster needed).
-- Only the OS1 Types are exercised for decoding: the three versions are
-- byte-identical (triple-duplicated), and the request-rendering parity
-- tests below pin the OS1/OS2/OS3 behavioural equivalence.
-- ========================================================================
describe "RollupId JSON" $ do
it "encodes RollupId as a bare JSON string" $ do
encode (OS1Types.RollupId "example_rollup")
`shouldBe` "\"example_rollup\""
it "decodes a bare JSON string into RollupId" $ do
decode "\"example_rollup\""
`shouldBe` Just (OS1Types.RollupId "example_rollup")
it "round-trips RollupId through encode -> decode" $ do
let original = OS1Types.RollupId "round-trip-id"
(decode . encode) original `shouldBe` Just original
describe "RollupMetricAgg JSON" $ do
it "encodes each constructor as a single-key object" $
forM_ metricAggTable $ \(agg, expected) ->
encode agg `shouldBe` expected
it "decodes each documented single-key object back" $
forM_ metricAggTable $ \(agg, wire) ->
decode wire `shouldBe` Just agg
it "decodes an unknown key via the Custom escape hatch" $ do
decode "{\"percentile\":{}}"
`shouldBe` Just (OS1Types.RollupMetricAggCustom "percentile")
it "rejects a multi-key object" $
decode "{\"avg\":{},\"sum\":{}}" `shouldSatisfy` (isNothing :: Maybe OS1Types.RollupMetricAgg -> Bool)
describe "RollupStatus JSON" $ do
it "encodes each documented status as the wire string" $
forM_ statusTable $ \(st, wire) ->
encode st `shouldBe` encode (String wire)
it "decodes each documented status back" $
forM_ statusTable $ \(st, wire) ->
decode (encode (String wire)) `shouldBe` Just st
it "decodes an unknown status via the Custom escape hatch" $
decode "\"paused\"" `shouldBe` Just (OS1Types.RollupStatusCustom "paused")
describe "RollupIntervalUnit JSON" $ do
it "encodes each documented unit as the wire string" $
forM_ intervalUnitTable $ \(u, wire) ->
encode u `shouldBe` encode (String wire)
it "decodes each documented unit back" $
forM_ intervalUnitTable $ \(u, wire) ->
decode (encode (String wire)) `shouldBe` Just u
describe "RollupDimension JSON" $ do
it "encodes a date_histogram dimension as a single-key object" $ do
let dh =
OS1Types.RollupDateHistogram
{ OS1Types.rollupDateHistogramSourceField = "ts",
OS1Types.rollupDateHistogramFixedInterval = Just "1h",
OS1Types.rollupDateHistogramCalendarInterval = Nothing,
OS1Types.rollupDateHistogramTimezone = Just "UTC",
OS1Types.rollupDateHistogramTargetField = Nothing
}
encode (OS1Types.RollupDimensionDateHistogram dh)
`shouldBe` encode (object ["date_histogram" .= dh])
it "encodes a terms dimension as a single-key object" $ do
let t = OS1Types.RollupTerms {OS1Types.rollupTermsSourceField = "cat", OS1Types.rollupTermsTargetField = Nothing}
encode (OS1Types.RollupDimensionTerms t)
`shouldBe` encode (object ["terms" .= t])
it "decodes a terms single-key object back" $ do
let t = OS1Types.RollupTerms {OS1Types.rollupTermsSourceField = "cat", OS1Types.rollupTermsTargetField = Nothing}
decode (encode (object ["terms" .= t]))
`shouldBe` Just (OS1Types.RollupDimensionTerms t)
describe "Rollup request body (wrapper)" $ do
it "decodes the documented create-request fixture" $ do
let decoded = decode sampleCreateRequestJson :: Maybe OS1Types.RollupWrapper
decoded `shouldSatisfy` isJust
let Just (OS1Types.RollupWrapper r) = decoded
OS1Types.rollupSourceIndex r `shouldBe` "nyc-taxi-data"
OS1Types.rollupTargetIndex r `shouldBe` "rollup-nyc-taxi-data"
OS1Types.rollupPageSize r `shouldBe` 200
length (OS1Types.rollupDimensions r) `shouldBe` 2
it "round-trips a Rollup through encode -> decode" $ do
let Just (OS1Types.RollupWrapper r') = decode (encode (OS1Types.RollupWrapper minimalOS1Rollup))
OS1Types.rollupSourceIndex r' `shouldBe` "src"
describe "RollupResponse JSON (dual seq/primary casing)" $ do
it "decodes the PUT response (snake_case _seq_no/_primary_term)" $ do
let Just resp = decode samplePutResponseJson :: Maybe OS1Types.RollupResponse
OS1Types.rollupResponseId resp `shouldBe` "example_rollup"
OS1Types.rollupResponseSeqNo resp `shouldBe` Just 1
OS1Types.rollupResponsePrimaryTerm resp `shouldBe` Just 1
OS1Types.rollupResponseInnerSchemaVersion (OS1Types.rollupResponseRollup resp)
`shouldBe` Just 17
it "decodes the GET response (camelCase _seqNo/_primaryTerm)" $ do
let Just resp = decode sampleGetResponseJson :: Maybe OS1Types.RollupResponse
OS1Types.rollupResponseId resp `shouldBe` "example_rollup"
OS1Types.rollupResponseSeqNo resp `shouldBe` Just 1
OS1Types.rollupResponsePrimaryTerm resp `shouldBe` Just 1
it "decodes a nullable metadata_id on the inner rollup" $ do
let Just resp = decode samplePutResponseJson :: Maybe OS1Types.RollupResponse
OS1Types.rollupResponseInnerMetadataId (OS1Types.rollupResponseRollup resp)
`shouldBe` Nothing
describe "ExplainRollupResponse JSON" $ do
it "decodes the not-yet-executed shape (both fields null)" $ do
let Just resp = decode sampleExplainNullJson :: Maybe OS1Types.ExplainRollupResponse
OS1Types.explainRollupResponseLookup "example_rollup" resp `shouldSatisfy` isJust
let Just entry = OS1Types.explainRollupResponseLookup "example_rollup" resp
OS1Types.explainRollupEntryMetadataId entry `shouldBe` Nothing
OS1Types.explainRollupEntryRollupMetadata entry `shouldBe` Nothing
it "decodes the executed shape with stats and a finished status" $ do
let Just resp = decode sampleExplainExecutedJson :: Maybe OS1Types.ExplainRollupResponse
let Just entry = OS1Types.explainRollupResponseLookup "example_rollup" resp
OS1Types.explainRollupEntryMetadataId entry `shouldBe` Just "GtWGlZwBm3bOohSSvi2r"
let Just md = OS1Types.explainRollupEntryRollupMetadata entry
OS1Types.rollupMetadataStatus md `shouldBe` OS1Types.RollupStatusFinished
OS1Types.rollupMetadataFailureReason md `shouldBe` Nothing
OS1Types.rollupStatsDocumentsProcessed <$> OS1Types.rollupMetadataStats md `shouldBe` Just 3
-- ========================================================================
-- Pure endpoint-shape tests (no live cluster needed). Each endpoint is
-- checked for method / path / query / body, and OS1/OS2/OS3 path parity
-- is asserted (the three modules are byte-identical; this guards against
-- refactors that silently diverge them).
-- ========================================================================
describe "createRollup endpoint shape" $ do
let rid = OS1Types.RollupId "my_rollup"
it "PUTs to /_plugins/_rollup/jobs/{id} with no query string" $ do
let req = OS1Requests.createRollup rid minimalOS1Rollup
bhRequestMethod req `shouldBe` "PUT"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs", "my_rollup"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "attaches the wrapped Rollup as the JSON body" $ do
let req = OS1Requests.createRollup rid minimalOS1Rollup
bhRequestBody req `shouldBe` Just (encode (OS1Types.RollupWrapper minimalOS1Rollup))
describe "createRollupWith optimistic concurrency" $ do
let rid = OS1Types.RollupId "my_rollup"
it "emits no query params when given Nothing" $ do
let req = OS1Requests.createRollupWith rid minimalOS1Rollup Nothing
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "emits if_seq_no and if_primary_term when given Just" $ do
let req = OS1Requests.createRollupWith rid minimalOS1Rollup (Just (5, 2))
sort (getRawEndpointQueries (bhRequestEndpoint req))
`shouldBe` sort [("if_seq_no", Just "5"), ("if_primary_term", Just "2")]
describe "getRollup endpoint shape" $
it "GETs /_plugins/_rollup/jobs/{id} with no body" $ do
let req = OS1Requests.getRollup (OS1Types.RollupId "my_rollup")
bhRequestMethod req `shouldBe` "GET"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs", "my_rollup"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
bhRequestBody req `shouldBe` Nothing
describe "getAllRollups endpoint shape" $
it "GETs /_plugins/_rollup/jobs with no body" $ do
let req = OS1Requests.getAllRollups
bhRequestMethod req `shouldBe` "GET"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs"]
bhRequestBody req `shouldBe` Nothing
describe "deleteRollup endpoint shape" $
it "DELETEs /_plugins/_rollup/jobs/{id} with no body" $ do
let req = OS1Requests.deleteRollup (OS1Types.RollupId "my_rollup")
bhRequestMethod req `shouldBe` "DELETE"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs", "my_rollup"]
bhRequestBody req `shouldBe` Nothing
describe "startRollup endpoint shape" $
it "POSTs /_plugins/_rollup/jobs/{id}/_start with no body" $ do
let req = OS1Requests.startRollup (OS1Types.RollupId "my_rollup")
bhRequestMethod req `shouldBe` "POST"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs", "my_rollup", "_start"]
bhRequestBody req `shouldBe` Nothing
describe "stopRollup endpoint shape" $
it "POSTs /_plugins/_rollup/jobs/{id}/_stop with no body" $ do
let req = OS1Requests.stopRollup (OS1Types.RollupId "my_rollup")
bhRequestMethod req `shouldBe` "POST"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs", "my_rollup", "_stop"]
bhRequestBody req `shouldBe` Nothing
describe "explainRollup endpoint shape" $
it "GETs /_plugins/_rollup/jobs/{id}/_explain with no body" $ do
let req = OS1Requests.explainRollup (OS1Types.RollupId "my_rollup")
bhRequestMethod req `shouldBe` "GET"
getRawEndpoint (bhRequestEndpoint req)
`shouldBe` ["_plugins", "_rollup", "jobs", "my_rollup", "_explain"]
bhRequestBody req `shouldBe` Nothing
describe "OS1/OS2/OS3 endpoint parity" $ do
let rid1 = OS1Types.RollupId "parity"
rid2 = OS2Types.RollupId "parity"
rid3 = OS3Types.RollupId "parity"
it "getRollup renders the same path across all three versions" $ do
let p1 = getRawEndpoint (bhRequestEndpoint (OS1Requests.getRollup rid1))
p2 = getRawEndpoint (bhRequestEndpoint (OS2Requests.getRollup rid2))
p3 = getRawEndpoint (bhRequestEndpoint (OS3Requests.getRollup rid3))
p1 `shouldBe` ["_plugins", "_rollup", "jobs", "parity"]
p1 `shouldBe` p2
p2 `shouldBe` p3
it "startRollup renders the same path across all three versions" $ do
let p1 = getRawEndpoint (bhRequestEndpoint (OS1Requests.startRollup rid1))
p2 = getRawEndpoint (bhRequestEndpoint (OS2Requests.startRollup rid2))
p3 = getRawEndpoint (bhRequestEndpoint (OS3Requests.startRollup rid3))
p1 `shouldBe` p2
p2 `shouldBe` p3
it "explainRollup renders the same path across all three versions" $ do
let p1 = getRawEndpoint (bhRequestEndpoint (OS1Requests.explainRollup rid1))
p2 = getRawEndpoint (bhRequestEndpoint (OS2Requests.explainRollup rid2))
p3 = getRawEndpoint (bhRequestEndpoint (OS3Requests.explainRollup rid3))
p1 `shouldBe` p2
p2 `shouldBe` p3
-- ========================================================================
-- Live integration tests (require the docker-compose cluster). One test
-- per backend, gated by osNOnlyIT. A timestamp-suffixed name avoids
-- collisions across runs; bracket_ guarantees the source + target
-- indices and the rollup job are torn down even on assertion failure.
-- The three versions are wire-identical, so the body is tripled (the
-- types are distinct at the Haskell level).
-- ========================================================================
describe "Index Rollups plugin live round-trip" $ do
os1It <- runIO os1OnlyIT
os1It "create -> get -> start -> explain -> stop -> delete (OpenSearch 1)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let src = T.pack ("bloodhound_test_rollup_os1_" <> suffix <> "_src")
tgt = T.pack ("bloodhound_test_rollup_os1_" <> suffix <> "_tgt")
rid = T.pack ("bloodhound_test_rollup_os1_" <> suffix <> "_job")
Right srcIdx = mkIndexName src
bracket_
(rollupSetupIndex srcIdx)
( do
void $ OS1Client.deleteRollup (OS1Types.RollupId rid)
void $ performBHRequest $ deleteIndex srcIdx
)
$ do
createResp <- OS1Client.createRollup (OS1Types.RollupId rid) (liveRollupOS1 src tgt)
liftIO $ case createResp of
Left e -> expectationFailure ("createRollup failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
getResp <- OS1Client.getRollup (OS1Types.RollupId rid)
liftIO $ assertRight "getRollup" getResp
startResp <- OS1Client.startRollup (OS1Types.RollupId rid)
liftIO $ assertRight "startRollup" startResp
explainResp <- OS1Client.explainRollup (OS1Types.RollupId rid)
liftIO $ assertRight "explainRollup" explainResp
stopResp <- OS1Client.stopRollup (OS1Types.RollupId rid)
liftIO $ assertRight "stopRollup" stopResp
os2It <- runIO os2OnlyIT
os2It "create -> get -> start -> explain -> stop -> delete (OpenSearch 2)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let src = T.pack ("bloodhound_test_rollup_os2_" <> suffix <> "_src")
tgt = T.pack ("bloodhound_test_rollup_os2_" <> suffix <> "_tgt")
rid = T.pack ("bloodhound_test_rollup_os2_" <> suffix <> "_job")
Right srcIdx = mkIndexName src
bracket_
(rollupSetupIndex srcIdx)
( do
void $ OS2Client.deleteRollup (OS2Types.RollupId rid)
void $ performBHRequest $ deleteIndex srcIdx
)
$ do
createResp <- OS2Client.createRollup (OS2Types.RollupId rid) (liveRollupOS2 src tgt)
liftIO $ case createResp of
Left e -> expectationFailure ("createRollup failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
getResp <- OS2Client.getRollup (OS2Types.RollupId rid)
liftIO $ assertRight "getRollup" getResp
startResp <- OS2Client.startRollup (OS2Types.RollupId rid)
liftIO $ assertRight "startRollup" startResp
explainResp <- OS2Client.explainRollup (OS2Types.RollupId rid)
liftIO $ assertRight "explainRollup" explainResp
stopResp <- OS2Client.stopRollup (OS2Types.RollupId rid)
liftIO $ assertRight "stopRollup" stopResp
os3It <- runIO os3OnlyIT
os3It "create -> get -> start -> explain -> stop -> delete (OpenSearch 3)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let src = T.pack ("bloodhound_test_rollup_os3_" <> suffix <> "_src")
tgt = T.pack ("bloodhound_test_rollup_os3_" <> suffix <> "_tgt")
rid = T.pack ("bloodhound_test_rollup_os3_" <> suffix <> "_job")
Right srcIdx = mkIndexName src
bracket_
(rollupSetupIndex srcIdx)
( do
void $ OS3Client.deleteRollup (OS3Types.RollupId rid)
void $ performBHRequest $ deleteIndex srcIdx
)
$ do
createResp <- OS3Client.createRollup (OS3Types.RollupId rid) (liveRollupOS3 src tgt)
liftIO $ case createResp of
Left e -> expectationFailure ("createRollup failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
getResp <- OS3Client.getRollup (OS3Types.RollupId rid)
liftIO $ assertRight "getRollup" getResp
startResp <- OS3Client.startRollup (OS3Types.RollupId rid)
liftIO $ assertRight "startRollup" startResp
explainResp <- OS3Client.explainRollup (OS3Types.RollupId rid)
liftIO $ assertRight "explainRollup" explainResp
stopResp <- OS3Client.stopRollup (OS3Types.RollupId rid)
liftIO $ assertRight "stopRollup" stopResp
-- | Build a typed OS1 'Rollup' for the live tests.
liveRollupOS1 :: T.Text -> T.Text -> OS1Types.Rollup
liveRollupOS1 src tgt =
OS1Types.Rollup
{ OS1Types.rollupSourceIndex = src,
OS1Types.rollupTargetIndex = tgt,
OS1Types.rollupTargetIndexSettings = Nothing,
OS1Types.rollupSchedule =
OS1Types.RollupSchedule $
OS1Types.RollupInterval
{ OS1Types.rollupIntervalPeriod = Just 1,
OS1Types.rollupIntervalUnit = Just OS1Types.RollupIntervalUnitMinutes,
OS1Types.rollupIntervalStartTime = Nothing,
OS1Types.rollupIntervalScheduleDelay = Nothing,
OS1Types.rollupIntervalCron = Nothing
},
OS1Types.rollupDescription = Just "bloodhound live test",
OS1Types.rollupEnabled = Just True,
OS1Types.rollupContinuous = Just False,
OS1Types.rollupPageSize = 200,
OS1Types.rollupDelay = Nothing,
OS1Types.rollupRoles = Nothing,
OS1Types.rollupRoutingField = Nothing,
OS1Types.rollupDimensions =
[ OS1Types.RollupDimensionDateHistogram $
OS1Types.RollupDateHistogram
{ OS1Types.rollupDateHistogramSourceField = "timestamp",
OS1Types.rollupDateHistogramFixedInterval = Just "1m",
OS1Types.rollupDateHistogramCalendarInterval = Nothing,
OS1Types.rollupDateHistogramTimezone = Just "UTC",
OS1Types.rollupDateHistogramTargetField = Nothing
}
],
OS1Types.rollupMetrics =
Just
[ OS1Types.RollupMetric
{ OS1Types.rollupMetricSourceField = "value",
OS1Types.rollupMetricMetrics = [OS1Types.RollupMetricAvg, OS1Types.RollupMetricSum]
}
],
OS1Types.rollupErrorNotification = Nothing
}
-- | Build a typed OS2 'Rollup' for the live tests.
liveRollupOS2 :: T.Text -> T.Text -> OS2Types.Rollup
liveRollupOS2 src tgt =
OS2Types.Rollup
{ OS2Types.rollupSourceIndex = src,
OS2Types.rollupTargetIndex = tgt,
OS2Types.rollupTargetIndexSettings = Nothing,
OS2Types.rollupSchedule =
OS2Types.RollupSchedule $
OS2Types.RollupInterval
{ OS2Types.rollupIntervalPeriod = Just 1,
OS2Types.rollupIntervalUnit = Just OS2Types.RollupIntervalUnitMinutes,
OS2Types.rollupIntervalStartTime = Nothing,
OS2Types.rollupIntervalScheduleDelay = Nothing,
OS2Types.rollupIntervalCron = Nothing
},
OS2Types.rollupDescription = Just "bloodhound live test",
OS2Types.rollupEnabled = Just True,
OS2Types.rollupContinuous = Just False,
OS2Types.rollupPageSize = 200,
OS2Types.rollupDelay = Nothing,
OS2Types.rollupRoles = Nothing,
OS2Types.rollupRoutingField = Nothing,
OS2Types.rollupDimensions =
[ OS2Types.RollupDimensionDateHistogram $
OS2Types.RollupDateHistogram
{ OS2Types.rollupDateHistogramSourceField = "timestamp",
OS2Types.rollupDateHistogramFixedInterval = Just "1m",
OS2Types.rollupDateHistogramCalendarInterval = Nothing,
OS2Types.rollupDateHistogramTimezone = Just "UTC",
OS2Types.rollupDateHistogramTargetField = Nothing
}
],
OS2Types.rollupMetrics =
Just
[ OS2Types.RollupMetric
{ OS2Types.rollupMetricSourceField = "value",
OS2Types.rollupMetricMetrics = [OS2Types.RollupMetricAvg, OS2Types.RollupMetricSum]
}
],
OS2Types.rollupErrorNotification = Nothing
}
-- | Build a typed OS3 'Rollup' for the live tests.
liveRollupOS3 :: T.Text -> T.Text -> OS3Types.Rollup
liveRollupOS3 src tgt =
OS3Types.Rollup
{ OS3Types.rollupSourceIndex = src,
OS3Types.rollupTargetIndex = tgt,
OS3Types.rollupTargetIndexSettings = Nothing,
OS3Types.rollupSchedule =
OS3Types.RollupSchedule $
OS3Types.RollupInterval
{ OS3Types.rollupIntervalPeriod = Just 1,
OS3Types.rollupIntervalUnit = Just OS3Types.RollupIntervalUnitMinutes,
OS3Types.rollupIntervalStartTime = Nothing,
OS3Types.rollupIntervalScheduleDelay = Nothing,
OS3Types.rollupIntervalCron = Nothing
},
OS3Types.rollupDescription = Just "bloodhound live test",
OS3Types.rollupEnabled = Just True,
OS3Types.rollupContinuous = Just False,
OS3Types.rollupPageSize = 200,
OS3Types.rollupDelay = Nothing,
OS3Types.rollupRoles = Nothing,
OS3Types.rollupRoutingField = Nothing,
OS3Types.rollupDimensions =
[ OS3Types.RollupDimensionDateHistogram $
OS3Types.RollupDateHistogram
{ OS3Types.rollupDateHistogramSourceField = "timestamp",
OS3Types.rollupDateHistogramFixedInterval = Just "1m",
OS3Types.rollupDateHistogramCalendarInterval = Nothing,
OS3Types.rollupDateHistogramTimezone = Just "UTC",
OS3Types.rollupDateHistogramTargetField = Nothing
}
],
OS3Types.rollupMetrics =
Just
[ OS3Types.RollupMetric
{ OS3Types.rollupMetricSourceField = "value",
OS3Types.rollupMetricMetrics = [OS3Types.RollupMetricAvg, OS3Types.RollupMetricSum]
}
],
OS3Types.rollupErrorNotification = Nothing
}
-- | Create the source index with one document. The @timestamp@ field
-- auto-maps to @date@ from the ISO-8601 string; @value@ auto-maps to a
-- numeric type. No explicit mapping is needed.
rollupSetupIndex :: IndexName -> BH IO ()
rollupSetupIndex indexName = do
_ <-
performBHRequest $
createIndex
(IndexSettings (ShardCount 1) (ReplicaCount 0) defaultIndexMappingsLimits)
indexName
nowMs <- liftIO $ round . (* 1000) <$> getPOSIXTime :: BH IO Integer
let ts = nowMs - 60000
iso = T.pack (iso8601FromEpochMs ts)
_ <-
performBHRequest $
indexDocument
indexName
defaultIndexDocumentSettings
(object ["timestamp" .= String iso, "value" .= Number 42])
(DocId "1")
_ <- performBHRequest $ refreshIndex indexName
pure ()
-- | Assert a 'ParsedEsResponse' is 'Right', failing with the plugin
-- error message otherwise.
assertRight :: T.Text -> Either EsError a -> IO ()
assertRight label r =
case r of
Left e ->
expectationFailure (T.unpack label <> " failed: " <> T.unpack (errorMessage e))
Right _ -> pure ()
-- | Render epoch-ms as an ISO-8601 UTC string (mirrors the AD spec helper).
iso8601FromEpochMs :: Integer -> String
iso8601FromEpochMs ms =
let secs = ms `div` 1000
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)