Revert "Add per-destination event filtering"

README.md

@@ -102,17 +102,6 @@ data_destinations:
       - sqlserver_source
       - file_source
 
-    # Optional. Deliver only records whose `column` value is in `values`;
-    # other records are skipped (the destination's cursor still advances).
-    # Omit to deliver every record. Records missing the column (or with a
-    # non-string value in it) are always delivered — a misconfigured column
-    # name degrades to full delivery, never to silently dropped records.
-    filter:
-      column: event_name
-      values:
-        - OrderCreated
-        - OrderUpdated
-
   # Send data to a file. One entry per line.
   - id: file_destination
     description: my projection 1

emulator.cabal

@@ -112,7 +112,6 @@ test-suite emulator-tests
     main-is: Tests.hs
     other-modules:
         Test.Config
-        Test.Projector
         Test.Queue
         Test.Connector
         Test.Connector.File

src/Ambar/Emulator.hs

@@ -123,7 +123,6 @@ emulate logger_ config env = do
         , p_destinationDescription = d_description dest
         , p_sources = sourceTopics
         , p_transport = transport
-        , p_filter = d_filter dest
         }
 
   withDestination dest act =

src/Ambar/Emulator/Config.hs

@@ -8,7 +8,6 @@ module Ambar.Emulator.Config
   , Source(..)
   , DataDestination(..)
   , Destination(..)
-  , DestinationFilter(..)
   , Port(..)
   )
   where
@@ -27,8 +26,6 @@ import qualified Data.Text as Text
 import Data.Text (Text)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
-import Data.Set (Set)
-import qualified Data.Set as Set
 import qualified Data.Yaml as Yaml
 
 import Ambar.Emulator.Connector.Poll (PollingInterval(..))
@@ -75,16 +72,6 @@ data DataDestination = DataDestination
   , d_sources :: [DataSource]
   , d_description :: Text
   , d_destination :: Destination
-  , d_filter :: Maybe DestinationFilter
-  }
-
--- | Server-side record filter for a destination. Only records whose
--- @f_column@ value is one of @f_values@ are delivered; everything else is
--- skipped (the consumer cursor still advances). A destination without a
--- filter receives every record, as before.
-data DestinationFilter = DestinationFilter
-  { f_column :: Text
-  , f_values :: Set Text
   }
 
 data Destination
@@ -200,7 +187,6 @@ parseDataDestination sourcesMap = Json.withObject "DataSource" $ \o -> do
           [ "Invalid data destination type: '" <> t <> "'."
           , "Expected one of: http-push, file."
           ]
-    d_filter <- o .:? "filter"
     return DataDestination{..}
     where
     parseHTTPPush o = do
@@ -211,14 +197,6 @@ parseDataDestination sourcesMap = Json.withObject "DataSource" $ \o -> do
 
     parseFile o = DestinationFile <$> (o .: "path")
 
-instance FromJSON DestinationFilter where
-  parseJSON = Json.withObject "DestinationFilter" $ \o -> do
-    f_column <- o .: "column"
-    values <- o .: "values"
-    when (null values) $
-      fail "filter.values must not be empty (omit the filter to deliver everything)"
-    return $ DestinationFilter f_column (Set.fromList values)
-
 parseEnvConfigFile :: FilePath -> IO EnvironmentConfig
 parseEnvConfigFile path = do
   bs <- BS.readFile path

src/Ambar/Emulator/Projector.hs

@@ -1,19 +1,14 @@
 module Ambar.Emulator.Projector
   ( Projection(..)
   , project
-  , filterVerdict
-  , FilterVerdict(..)
   , Message(..)
   , Payload(..)
   ) where
 
 {-| A projector reads messages from multiple queues, applies a filter to the
 stream and submits passing messages to a single data destination.
 
-A destination with no filter receives every record. A destination with a
-filter receives only records whose filter-column value is in the allowed
-set; filtered-out records are committed without being sent, so the
-consumer cursor always advances.
+We do not implement filters for now.
 -}
 
 import qualified Data.Aeson as Json
@@ -26,14 +21,10 @@ import Data.Text (Text)
 import qualified Data.Text.Encoding as Text
 import qualified Data.Text as Text
 import Control.Concurrent.Async (forConcurrently_)
-import Control.Monad (when)
 import Control.Monad.Extra (whileM)
-import Data.IORef (newIORef, atomicModifyIORef')
 import GHC.Generics (Generic)
 
-import qualified Data.Set as Set
-
-import Ambar.Emulator.Config (Id(..), DataDestination, DataSource(..), Source(..), DestinationFilter(..))
+import Ambar.Emulator.Config (Id(..), DataDestination, DataSource(..), Source(..))
 import Ambar.Emulator.Queue.Topic (Topic, ReadError(..), PartitionCount(..))
 import qualified Ambar.Emulator.Queue.Topic as Topic
 import Ambar.Emulator.Connector.MicrosoftSQLServer (SQLServer(..))
@@ -53,7 +44,6 @@ data Projection = Projection
   , p_destinationDescription :: Text
   , p_sources :: [(DataSource, Topic)]
   , p_transport :: Some Transport
-  , p_filter :: Maybe DestinationFilter
   }
 
 -- | A record enriched with more information to send to the client.
@@ -72,51 +62,31 @@ newtype Payload = Payload Json.Value
   deriving newtype (ToJSON, FromJSON)
 
 project :: SimpleLogger -> Projection -> IO ()
-project logger_ Projection{..} = do
-  -- One warning per destination per emulator run when the filter fails
-  -- open (reviewer note on PR #56): a typo'd filter column silently
-  -- restores full delivery, which looks exactly like a normal match.
-  -- Warning on every record would flood the log at full event rate, so
-  -- the first fail-open claims this flag and later ones stay quiet.
-  warnedFailOpen <- newIORef False
-  let warnFailOpenOnce logger reason = do
-        firstWarn <- atomicModifyIORef' warnedFailOpen (\claimed -> (True, not claimed))
-        when firstWarn $
-          logWarn logger $
-            "filter fail-open: " <> reason
-            <> ". Delivering the record; the destination filter is not being applied."
-            <> " Further fail-open warnings for this destination are suppressed."
-  forConcurrently_ p_sources (projectSource warnFailOpenOnce)
+project logger_ Projection{..} =
+  forConcurrently_ p_sources projectSource
   where
-  projectSource warnFailOpenOnce (source, topic) =
+  projectSource (source, topic) =
     -- one consumer per partition
     Topic.withConsumers topic group pcount $ \consumers ->
     forConcurrently_ consumers $ \consumer ->
-    whileM $ consume warnFailOpenOnce logger consumer source
+    whileM $ consume logger consumer source
     where
       PartitionCount pcount = Topic.partitionCount topic
       logger =
         annotate ("src: " <> unId (s_id source)) $
         annotate ("dst: " <> unId  p_destination)
         logger_
 
-  consume warnFailOpenOnce logger consumer source = do
+  consume logger consumer source = do
     r <- Topic.read consumer
     case r of
       Left EndOfPartition -> return False
       Left err -> fatal logger (show err)
       Right (bs, meta) -> do
         record <- decode logger bs
         let logger' = annotate (relevantFields (s_source source) record) logger
-            send = do
-              retrying logger' $ Transport.sendJSON p_transport (toMsg source record)
-              logInfo logger' ("sent." :: Text)
-        case filterVerdict p_filter record of
-          Matched -> send
-          Skipped -> logInfo logger' ("filtered." :: Text)
-          FailedOpen reason -> do
-            warnFailOpenOnce logger' reason
-            send
+        retrying logger' $ Transport.sendJSON p_transport (toMsg source record)
+        logInfo logger' ("sent." :: Text)
         Topic.commit consumer meta
         return True
 
@@ -139,33 +109,6 @@ project logger_ Projection{..} = do
       fatal logger $ "decoding error: " <> err <> "\nraw: " <> raw
     Right v -> return v
 
--- | The filter's decision for a record.
---
--- 'FailedOpen' means the record is DELIVERED even though the filter could
--- not be applied (missing column, null or non-string value, non-object
--- record): a misconfigured column name must degrade to full delivery (the
--- pre-filter behaviour), never to silently dropped records. The reason is
--- carried so the projector can warn that the filter is not doing its job.
-data FilterVerdict
-  = Matched            -- ^ deliver: no filter, or the column value is allowed
-  | Skipped            -- ^ commit without delivering
-  | FailedOpen Text    -- ^ deliver, but the filter could not be applied
-  deriving (Eq, Show)
-
-filterVerdict :: Maybe DestinationFilter -> Payload -> FilterVerdict
-filterVerdict Nothing _ = Matched
-filterVerdict (Just DestinationFilter{..}) (Payload value) =
-  case value of
-    Json.Object o ->
-      case KeyMap.lookup (fromString $ Text.unpack f_column) o of
-        Just (Json.String v)
-          | Set.member v f_values -> Matched
-          | otherwise -> Skipped
-        Just Json.Null -> FailedOpen $ "filter column '" <> f_column <> "' is null"
-        Just _ -> FailedOpen $ "filter column '" <> f_column <> "' has a non-string value"
-        Nothing -> FailedOpen $ "filter column '" <> f_column <> "' is missing from the record"
-    _ -> FailedOpen "record is not a JSON object"
-
 -- | Fields to print when a record is sent.
 relevantFields :: Source -> Payload -> Text
 relevantFields source (Payload value) = renderPretty $