Add network policy summary endpoint for the prometheus metrics

- Introduced new 'network_policies/summary' endpoint to the project's
  security group.
- Introduced new Prometheus query to fetch necessary stats.

ee/app/controllers/projects/security/network_policies_controller.rb

+# frozen_string_literal: true
+
+module Projects
+  module Security
+    class NetworkPoliciesController < Projects::ApplicationController
+      POLLING_INTERVAL = 5_000
+
+      before_action :authorize_read_network_policies!
+      before_action :set_polling_interval, only: [:summary]
+
+      def summary
+        return not_found unless environment.has_metrics?
+
+        adapter = environment.prometheus_adapter
+        return not_found unless adapter.can_query?
+
+        result = adapter.query(
+          :packet_flow, environment.deployment_namespace,
+          interval: params[:interval] || "minute",
+          from:     (Time.parse(params[:from]) rescue 1.hour.ago),
+          to:       (Time.parse(params[:to]) rescue Time.now)
+        )
+
+        respond_to do |format|
+          format.json do
+            status = result[:success] ? :ok : :bad_request
+            render status: status, json: result[:data]
+          end
+        end
+      end
+
+      private
+
+      def environment
+        @environment ||= project.environments.find(params[:environment_id])
+      end
+
+      def set_polling_interval
+        Gitlab::PollingInterval.set_header(response, interval: POLLING_INTERVAL)
+      end
+
+      def authorize_read_network_policies!
+        render_403 unless can?(current_user, :read_threat_monitoring, project)
+      end
+    end
+  end
+end

ee/config/routes/project.rb

@@ -78,6 +78,10 @@ constraints(::Constraints::ProjectUrlConstrainer.new) do
             get :summary, on: :collection
           end
 
+          resource :network_policies, only: [] do
+            get :summary, on: :collection
+          end
+
           resources :dashboard, only: [:index], controller: :dashboard
           resource :configuration, only: [:show], controller: :configuration
           resource :discover, only: [:show], controller: :discover

ee/lib/gitlab/prometheus/queries/packet_flow_query.rb

+# frozen_string_literal: true
+
+module Gitlab
+  module Prometheus
+    module Queries
+      class PacketFlowQuery < BaseQuery
+        FORWARDED = "FORWARDED".freeze
+        DROPPED   = "DROPPED".freeze
+
+        def query(namespace, interval: "hour", from: 1.day.ago, to: Time.now)
+          rate_interval = to_prometheus_interval(interval)
+          increase_interval = (to - from).to_i
+
+          rate_query = sum_by_verdict(
+            rate_query(%{destination="#{namespace}"}, rate_interval),
+            rate_query(%{source="#{namespace}"}, rate_interval)
+          )
+          total_query = sum_by_verdict(
+            increase_query(%{destination="#{namespace}"}, increase_interval),
+            increase_query(%{source="#{namespace}"}, increase_interval)
+          )
+
+          {
+            ops_rate:  transform_rate_result(client_query_range(rate_query, start: from, stop: to)),
+            ops_total: transform_sum_result(client_query(total_query, time: to))
+          }
+        end
+
+        private
+
+        def sum_by_verdict(vec1, vec2)
+          %{sum by(verdict) (#{vec1} or on(source,destination,verdict) #{vec2})}
+        end
+
+        def rate_query(selector, interval)
+          %{rate(hubble_flows_processed_total{#{selector}}[#{interval}])}
+        end
+
+        def increase_query(selector, interval)
+          %{increase(hubble_flows_processed_total{#{selector}}[#{interval}s])}
+        end
+
+        # Returns rate of packets and dropped packets from a range vector:
+        # [
+        #  {"metric"=>{"verdict"=>"FORWARDED"}, "values"=>[[1582231596.64, "73772.43143284984"]]},
+        #  {"metric"=>{"verdict"=>"DROPPED"}, "values"=>[[1582231596.64, "5.002730665588791"]]}
+        # ]
+        def transform_rate_result(vector)
+          values = vector.each_with_object({ DROPPED => [], FORWARDED => [] }) do |val, acc|
+            acc[val["metric"].first.second] = val["values"].map { |item| [item.first, item.last.to_f] }
+          end
+
+          forwards = values[FORWARDED]
+          drops = values[DROPPED]
+
+          { total: sum_vectors(drops, forwards), drops: drops }
+        end
+
+        # Returns total amount of packets and dropped packets from an instant vector:
+        # [
+        #  {"metric"=>{"verdict"=>"FORWARDED"}, "value"=>[1582231596.64, "73772.43143284984"]},
+        #  {"metric"=>{"verdict"=>"DROPPED"}, "value"=>[1582231596.64, "5.002730665588791"]}
+        # ]
+        def transform_sum_result(vector)
+          values = vector.each_with_object({ DROPPED => 0, FORWARDED => 0 }) do |val, acc|
+            acc[val["metric"].first.second] = val["value"].last.to_i
+          end
+
+          forwards = values[FORWARDED]
+          drops = values[DROPPED]
+
+          { total: forwards + drops, drops: drops }
+        end
+
+        def to_prometheus_interval(interval)
+          case interval
+          when "hour" then "1h"
+          when "day" then "1d"
+          else
+            "5m"
+          end
+        end
+
+        # Sums 2 vectors in format [[ts1, val1 + val2]]
+        # Expects vectors to be equal length. One of the vectors can be nil
+        def sum_vectors(vec1, vec2)
+          return vec1 if vec2.empty?
+          return vec2 if vec1.empty?
+
+          acc = []
+          vec1.zip(vec2) { |v1, v2| acc << [v1.first, v1.second + v2.second] }
+          acc
+        end
+      end
+    end
+  end
+end

ee/spec/controllers/projects/security/network_policies_controller_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Projects::Security::NetworkPoliciesController do
+  let_it_be(:group) { create(:group) }
+  let_it_be(:user) { create(:user) }
+
+  let_it_be(:project) { create(:project, :public, :repository, group: group) }
+  let_it_be(:environment) { create(:environment, :with_review_app, project: project) }
+
+  let_it_be(:action_params) { { project_id: project, namespace_id: project.namespace, environment_id: environment } }
+
+  describe 'GET #summary' do
+    subject { get :summary, params: action_params, format: :json }
+
+    let_it_be(:kubernetes_namespace) { environment.deployment_namespace }
+
+    before do
+      stub_licensed_features(threat_monitoring: true)
+
+      sign_in(user)
+    end
+
+    context 'with authorized user' do
+      before do
+        group.add_developer(user)
+      end
+
+      context 'with prometheus configured' do
+        let(:adapter) { double("configured?" => true, "can_query?" => true) }
+
+        before do
+          allow_next_instance_of(Gitlab::Prometheus::Adapter) do |instance|
+            allow(instance).to receive(:prometheus_adapter) { adapter }
+          end
+        end
+
+        it 'returns network policies summary' do
+          Timecop.freeze do
+            expect(adapter).to(
+              receive(:query)
+                .with(:packet_flow, kubernetes_namespace, interval: "minute", from: 1.hour.ago, to: Time.now)
+                .and_return({ success: true, data: { ops_rate: [[Time.at(0).to_i, 10]], ops_total: 10 } })
+            )
+            subject
+          end
+
+          expect(response).to have_gitlab_http_status(:ok)
+          expect(json_response['ops_total']).to equal(10)
+          expect(json_response['ops_rate']).to eq([[0, 10]])
+        end
+
+        context 'with additional parameters' do
+          let(:action_params) do
+            {
+              project_id: project, namespace_id: project.namespace, environment_id: environment,
+              interval:  "day", from: Time.at(0), to: Time.at(100)
+            }
+          end
+
+          it 'queries with requested arguments' do
+            expect(adapter).to(
+              receive(:query)
+                .with(:packet_flow, kubernetes_namespace, interval: "day", from: Time.at(0), to: Time.at(100))
+                .and_return({ success: true, data: {} })
+            )
+            subject
+          end
+        end
+
+        context 'with invalid Time range' do
+          let(:action_params) do
+            {
+              project_id: project, namespace_id: project.namespace, environment_id: environment,
+              from: "not a time", to: "not a time"
+            }
+          end
+
+          it 'queries with default arguments' do
+            Timecop.freeze do
+              expect(adapter).to(
+                receive(:query)
+                  .with(:packet_flow, kubernetes_namespace, interval: "minute", from: 1.hour.ago, to: Time.now)
+                  .and_return({ success: true, data: {} })
+              )
+              subject
+            end
+          end
+        end
+      end
+
+      context 'without prometheus configured' do
+        it 'returns not found' do
+          subject
+
+          expect(response).to have_gitlab_http_status(:not_found)
+        end
+      end
+
+      it 'sets a polling interval header' do
+        subject
+
+        expect(response.headers['Poll-Interval']).to eq('5000')
+      end
+    end
+
+    context 'with unauthorized user' do
+      it 'returns unauthorized' do
+        subject
+
+        expect(response).to have_gitlab_http_status(:forbidden)
+      end
+    end
+  end
+end

ee/spec/lib/gitlab/prometheus/queries/packet_flow_query_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe Gitlab::Prometheus::Queries::PacketFlowQuery do
+  let(:namespace) { 'query-12345678-production' }
+  let(:query_range_response) { [] }
+  let(:query_response) { [] }
+  let(:client) { double('prometheus_client', query: query_response, query_range: query_range_response) }
+
+  subject { described_class.new(client) }
+
+  context 'metrics' do
+    let(:query_range_response) do
+      [
+        { "metric" => { "verdict" => "FORWARDED" }, "values" => [[1582231596.64, "73772.43143284984"]] },
+        { "metric" => { "verdict" => "DROPPED" }, "values" => [[1582231596.64, "5.002730665588791"]] }
+      ]
+    end
+    let(:query_response) do
+      [
+        { "metric" => { "verdict" => "FORWARDED" }, "value" => [1582231596.64, "73772.43143284984"] },
+        { "metric" => { "verdict" => "DROPPED" }, "value" => [1582231596.64, "5.002730665588791"] }
+      ]
+    end
+    let(:result) { subject.query(namespace) }
+
+    it 'returns ops_rate metric' do
+      expect(result[:ops_rate]).to(
+        eq(
+          {
+            total: [[1582231596.64, 73777.43416351543]],
+            drops: [[1582231596.64, 5.002730665588791]]
+          }
+        )
+      )
+    end
+
+    it 'returns ops_total metric' do
+      expect(result[:ops_total]).to eq({ total: 73777, drops: 5 })
+    end
+  end
+
+  context 'query' do
+    it 'sends ops_rate prometheus query' do
+      query = 'sum by(verdict) (' \
+              'rate(hubble_flows_processed_total{destination="query-12345678-production"}[1h])' \
+              ' or on(source,destination,verdict) ' \
+              'rate(hubble_flows_processed_total{source="query-12345678-production"}[1h]))'
+      expect(client).to receive(:query_range).with(query, any_args)
+      subject.query(namespace)
+    end
+
+    it 'sends ops_total prometheus query' do
+      query = 'sum by(verdict) (' \
+              'increase(hubble_flows_processed_total{destination="query-12345678-production"}[86400s])' \
+              ' or on(source,destination,verdict) ' \
+              'increase(hubble_flows_processed_total{source="query-12345678-production"}[86400s]))'
+      expect(client).to receive(:query).with(query, any_args)
+      subject.query(namespace)
+    end
+  end
+
+  context 'ops_rate intervals' do
+    { "minute" => "5m", "hour" => "1h", "day" => "1d" }.each do |interval, value|
+      context "#{interval} interval" do
+        it 'uses correct interval' do
+          query = 'sum by(verdict) (' \
+                  "rate(hubble_flows_processed_total{destination=\"query-12345678-production\"}[#{value}])" \
+                  ' or on(source,destination,verdict) ' \
+                  "rate(hubble_flows_processed_total{source=\"query-12345678-production\"}[#{value}]))"
+          expect(client).to receive(:query_range).with(query, any_args)
+          subject.query(namespace, interval: interval)
+        end
+      end
+    end
+  end
+
+  context 'time range' do
+    let(:from) { Time.at(0) }
+    let(:to) { Time.at(100) }
+
+    context 'ops_rate query' do
+      it 'sets query time range' do
+        expect(client).to receive(:query_range).with(anything, start: from, stop: to)
+        subject.query(namespace, from: from, to: to)
+      end
+    end
+
+    context 'ops_total query' do
+      it 'sets query time range and interval' do
+        query = 'sum by(verdict) (' \
+                'increase(hubble_flows_processed_total{destination="query-12345678-production"}[100s])' \
+                ' or on(source,destination,verdict) ' \
+                'increase(hubble_flows_processed_total{source="query-12345678-production"}[100s]))'
+        expect(client).to receive(:query).with(query, time: to)
+        subject.query(namespace, from: from, to: to)
+      end
+    end
+  end
+end