     (when-not valid?
       (if-not @error-reported?
         (do
-          (swap! error-paths conj path)
+          (swap! mstore-error-paths conj path)
           (wlog/error err)
           (swap! error-reported? (fn [_] true)))
         (log/debug err)))
     valid?))
 
-(defn- validate-points
-  "Validate points."
-  [mstore options tenant rollup period retention path series]
-  (let [series (whisper/sort-series series)
-        from (first (first series))
-        to (first (last series))
-        min-ttl (:min-ttl options default-min-ttl)
-        mstore-series (mstore/fetch-series mstore tenant rollup period path
-                                           from to)
-        error-reported? (atom false)
-        validators [[#(not= %2 nil) "Point not found"]
-                    [#(sequential? %2) "Value is not an array"]
-                    [#(= (count %2) 1) "Array size is not equal to 1"]
-                    [#(= %1 (first %2)) "Values are not equal"]]
-        error-paths (:error-paths (:data-mstore-stats options))]
-    (doseq [point series]
-      (let [time (first point)
-            w-value (last point)
-            ttl (calc-ttl time retention)]
-        (when (> ttl min-ttl)
-          (log-point "Validating point" rollup period path time w-value ttl)
-          (let [s-value (get mstore-series time)]
-            (every? #(validate-value rollup period path time w-value s-value
-                                     (first %) (second %) error-reported?
-                                     error-paths)
-                    validators)))))))
+(defn- validator
+  "Validator."
+  [mstore pstore options tenant]
+  (let [min-ttl (:min-ttl options default-min-ttl)]
+    (reify
+      Processor
+      (process-metrics [this rollup period retention path series]
+        (let [series (whisper/sort-series series)
+              from (first (first series))
+              to (first (last series))
+              mstore-series (mstore/fetch-series mstore tenant rollup period path
+                                                 from to)
+              error-reported? (atom false)
+              validators [[#(not= %2 nil) "Point not found"]
+                          [#(sequential? %2) "Value is not an array"]
+                          [#(= (count %2) 1) "Array size is not equal to 1"]
+                          [#(= %1 (first %2)) "Values are not equal"]]]
+          (doseq [point series]
+            (let [time (first point)
+                  w-value (last point)
+                  ttl (calc-ttl time retention)]
+              (when (> ttl min-ttl)
+                (log-point "Validating point" rollup period path time w-value ttl)
+                (let [s-value (get mstore-series time)]
+                  (every? #(validate-value rollup period path time w-value
+                                           s-value (first %) (second %)
+                                           error-reported? mstore-error-paths)
+                          validators)))))))
+      (process-path [this path]
+        (when-not (pstore/exist? pstore tenant path)
+          (swap! pstore-error-paths conj path)
+          (wlog/error "Path not found in the path store: " path))))))
 
 (defn- process-archive
   "Process an archive."
-  [ra-file file archive mstore tenant path from to retention options points-fn]
+  [ra-file file archive mstore tenant path from to retention options processor]
   (let [rollup (:seconds-per-point archive)
         points (:points archive)
         period (if retention (/ retention rollup) points)
         retention (if retention retention (* rollup points))
         data (whisper/fetch-archive-seq-ra ra-file file archive from to)
         series (whisper/remove-nulls (:series data))]
-    (when points-fn
-      (points-fn mstore options tenant rollup period retention path series))))
-
-(defn- migrate-path
-  "Migrate a path."
-  [pstore tenant path options]
-  (let [run (:run options false)]
-    (when run
-      (pstore/insert pstore tenant path))))
-
-(defn- validate-path
-  "Validate a path."
-  [pstore tenant path options]
-  (let [error-paths (:error-paths (:data-pstore-stats options))]
-    (when-not (pstore/exist? pstore tenant path)
-      (swap! error-paths conj path)
-      (wlog/error "Path not found in the path store: " path))))
+    (process-metrics processor rollup period retention path series)))
 
 (defn- process-file
   "Process a file."
-  [file dir mstore pstore tenant from to options path-fn points-fn]
+  [file dir mstore pstore tenant from to options processor]
   (let [path (whisper/file-to-name file dir)
         mstore-processed (:processed (:data-mstore-stats options))
         pstore-processed (:processed (:data-pstore-stats options))]
-    (when (and pstore path-fn)
+    (when pstore
       (swap! pstore-processed inc)
-      (path-fn pstore tenant path options))
+      (process-path processor path))
     (when mstore
       (let [ra-file (RandomAccessFile. file "r")
             rollups (:rollups options)]

   (format (str "%s: rollup: %s, period: %s, path: %s, time: %s, "
                "value: %s, ttl: %s") msg rollup period path time value ttl)))

(defn- migrate-points
  "Migrate points."
  [mstore options tenant rollup period retention path series]
  (let [min-ttl (:min-ttl options default-min-ttl)
        run (:run options false)]
    (doseq [point series]
      (let [time (first point)
            value (last point)
            ttl (calc-ttl time retention)]
        (when (> ttl min-ttl)
          (log-point "Migrating point" rollup period path time value ttl)
          (when run
            (mstore/insert mstore tenant rollup period path time value
                           ttl)))))))

(defn- validate-value
  "Validate a value."

;;     (assoc merged :error-paths (clojure.set/union (:error-paths total)
;;                                                   (:error-paths store)))))

(defn- show-stats
  "Show metric store stats."
  [title data-stats store-stats]
