Закрытый проект Римана

Question

Я пытаюсь сделать пользовательскую настройку, по-видимому, простой, используя Римана и Коллектда. В основном я хотел бы рассчитать соотношение между двумя потоками. Чтобы сделать это, я попробовал что-то вроде (как в предложении проекта Rieamann API здесь):

(project [(service "cahe-miss")
      (service "cache-all")]
  (smap folds/quotient
    (with :service "ratio"
    index)))

Что, видимо, работает, но через некоторое время я заметил некоторые результаты, где рассчитывается мисс. После отладки журнала я закончил со следующей конфигурацией, чтобы увидеть, что происходит, и набрать значения:

(project [(service "cache-miss")
          (service "cache-all")]
  (fn [[miss all]]
    (if (or (nil? miss) (nil? all)) 
      (do nil) 
      (do (where (= (:time miss) (:time all))
        ;to print time marks
        (println (:time all)) 
        (println (:time miss))
        ; to distinguish easily each event
        (println "NEW LINE") 
      )) 
    )
  )
)

Мое удивление заключается в том, что каждый раз, когда я получаю новые данные из collectd (каждые 10 секунд), созданная мной функция выполняется дважды, как повторное использование предыдущих неиспользуемых данных, и, более того, похоже, что меня не волнует ограничение равенства времени в (где (=: время....). Проблема в том, что я делю метрики с другой отметкой времени. Ниже приведен вывод предыдущего кода:

1445606294
1445606294
NEW LINE -- First time I get data
1445606304
1445606294
NEW LINE
1445606304
1445606304
NEW LINE -- Second time I get data
1445606314
1445606304
NEW LINE
1445606314
1445606314
NEW LINE -- Third time I get data

Кто-нибудь может подсказать, как отформатировать данные, как я ожидал? Я предполагаю, что есть кое-что, чего я не понимаю в функции "проекта". Или что-то, связанное с тем, как поступающие данные обрабатываются в riemann.

Заранее спасибо!

обновленный

Мне удалось решить мою проблему, но я до сих пор не имею четкого представления о том, как она работает, однако мне удалось это сделать. Прямо сейчас я получаю два разных потока из плагина taild tail (из логов nginx), и мне удалось сделать соотношение между ними следующим образом:

(where (or (service "nginx/counter-cacheHit") (service "nginx/counter-cacheAll"))
    (coalesce
        (smap folds/quotient (with :service "cacheHit" (scale (* 1 100) index)))))

Я проверил это широко и до сих пор он дает правильные результаты. Однако я до сих пор не понимаю нескольких вещей... Во-первых, как получается, что объединение возвращает данные только после обработки обоих событий. Collectd отправляет события обоих потоков каждые две секунды с одинаковой отметкой времени, используя "проект" вместо "объединить", что приводит к двум различным выполнениям smap каждые две секунды (по одному для каждого события), однако объединение приводит только к одному выполнению smap с двумя событиями с одинаковой отметкой времени, что именно то, что я хотел.

Наконец, я не знаю, какой критерий выбрать, какой числитель и знаменатель. Является ли это из-за "или" в разделе "где"?

Во всяком случае, с какой-то черной магией, но мне удалось решить мою проблему;^)

Спасибо вам всем!

Answer 1

Принимая отношения между потоками, которые, где движение с различными скоростями, не работало для меня. С тех пор я остановился на расчете коэффициентов и ставок в течение фиксированного интервала времени или временного интервала. Таким образом, вы получаете последовательный снимок событий во временном блоке и рассчитываете его. Вот некоторый проверенный код для сравнения скорости, с которой служба получает события, и скорости, с которой она пересылает события:

(moving-time-window 30 ;; seconds
  (smap (fn [events]
          (let [in (or (->> events
                            (filter #(= (:service %) "event-received"))
                            count)
                       0)
                out (or (->> events
                             (filter #(= (:service %) "event-sent"))
                             count)
                        0)
                flow-rate (float (if (> in 0) (/ out in) 0))]
            {:service "flow rate"
             :metric flow-rate
             :host "All"
             :state (if (< flow-rate 0.99) "WARNING" "OK")
             :time (:time (last events))
             :ttl default-interval}))
        (tag ["some" "tags" "here"] index)
        (where (and
                (< (:metric event) 0.9)
                (= (:environment event) "production"))
               (throttle 1 3600 send-to-slack))))

Это принимает фиксированное окно событий, вычисляет отношение для этого блока и генерирует событие, содержащее это отношение в качестве метрики. тогда, если метрика плохая, это вызывает у меня слабину.