Added support for cadvisor metrics

Rakefile

@@ -24,6 +24,6 @@ namespace :docker do
     raise "Argument `tag` was not provided." unless args.tag
 
     cp Dir['pkg/fluent-plugin-kubernetes-metrics-*.gem'], 'docker/'
-    sh "docker build -t splunk/connect-for-kubernetes:#{args.tag} ./docker"
+    sh "docker build  --no-cache -t splunk/connect-for-kubernetes:#{args.tag} ./docker"
   end
 end

docker/Dockerfile

@@ -1,40 +1,43 @@
-FROM centos:7.5.1804
+FROM alpine:3.8
 
 LABEL maintainer="Don Tregonning <dtregonning@splunk.com>, Chaitanya Phalak <cphalak@splunk.com>"
 LABEL Description="Splunk Connect for Kubernetes docker image" Vendor="Splunk Inc." Version="1.1.Alpha"
 
+ENV DUMB_INIT_VERSION=1.2.1
+ENV SU_EXEC_VERSION=0.2
+
 COPY *.gem /tmp/
-RUN curl -L -o /tmp/epel-release-7-11.noarch.rpm http://dl.fedoraproject.org/pub/epel/7/x86_64/Packages/e/epel-release-7-11.noarch.rpm \
- && rpm -Uvh /tmp/epel-release*rpm \
- && yum -y update \
- && yum install -y jemalloc jq-devel \
- && yum -y groupinstall "Development Tools" \
- && yum -y install libxslt-devel libyaml-devel libxml2-devel gdbm-devel libffi-devel zlib-devel openssl-devel libyaml-devel readline-devel curl-devel openssl-devel pcre-devel git memcached-devel valgrind-devel mysql-devel ImageMagick-devel ImageMagick \
- && cd /usr/local/src \
- && curl -L -o ruby-2.5.1.tar.gz https://cache.ruby-lang.org/pub/ruby/2.5/ruby-2.5.1.tar.gz \
- && tar zxvf ruby-2.5.1.tar.gz \
- && cd ruby-2.5.1 \
- && ./configure \
- && make \
- && make install \
- && gem install -N fluentd:1.3.0 \
-                   fluent-plugin-systemd:1.0.1 \
-                   fluent-plugin-concat:2.3.0 \
-                   fluent-plugin-prometheus:1.2.1 \
-                   fluent-plugin-jq:0.5.1 \
-                   fluent-plugin-record-modifier:1.1.0 \
-                   fluent-plugin-splunk-hec:1.0.1 \
-                   oj:3.7.1 \
+
+ARG DEBIAN_FRONTEND=noninteractive
+# Do not split this into multiple RUN!
+# Docker creates a layer for every RUN-Statement
+# therefore an 'apk delete' has no effect
+RUN apk update \
+ && apk upgrade \
+ && apk add --no-cache \
+        ca-certificates \
+ && update-ca-certificates \
+ && apk add --no-cache \
+        ruby ruby-irb ruby-etc ruby-webrick ruby-json \
+        su-exec==${SU_EXEC_VERSION}-r0 \
+        dumb-init==${DUMB_INIT_VERSION}-r0 \
+ && apk add --no-cache --virtual .build-deps \
+        build-base \
+        ruby-dev wget gnupg \
+ && echo 'gem: --no-document' >> /etc/gemrc \
+ && gem install -N \
+         fluentd:1.3.1 \
+         fluent-plugin-record-modifier:1.1.0 \
+         fluent-plugin-splunk-hec:1.0.1 \
+         oj:3.7.4 \
+         multi_json:1.13.1 \
+         bigdecimal:1.3.5 \
  && gem install -N /tmp/*.gem \
- && curl -L -o /usr/bin/dumb-init https://github.com/Yelp/dumb-init/releases/download/v1.2.2/dumb-init_1.2.2_amd64 \
- && chmod +x /usr/bin/dumb-init \
- && curl -L -o /usr/bin/gosu https://github.com/tianon/gosu/releases/download/1.10/gosu-amd64 \
- && chmod +x /usr/bin/gosu \
- && rm -rf /tmp/* /var/tmp/* $GEM_HOME/cache/*.gem \
- && mkdir -p /fluentd/{etc,log,plugins}
+ && apk del .build-deps \
+ && rm -rf /var/cache/apk/* \
+ && rm -rf /tmp/* /var/tmp/* /usr/lib/ruby/gems/*/cache/*.gem
 
-# from `repoquery -l jemalloc`
-ENV LD_PRELOAD="/usr/lib64/libjemalloc.so.1" \
+ENV LD_PRELOAD="" \
     FLUENTD_CONF="fluent.conf" \
     FLUENTD_OPT="" \
     DUMB_INIT_SETSID=0

docker/entrypoint.sh

@@ -9,8 +9,7 @@ if [ $? -eq 0 ]; then
 fi
 
 # (re)add the fluent user with $FLUENT_UID
-useradd -u ${uid} -o -c "" -m fluent
-export HOME=/home/fluent
+adduser -D -g '' -u ${uid} -h /home/fluent fluent
 
 #source vars if file exists
 DEFAULT=/etc/default/fluentd
@@ -25,4 +24,4 @@ fi
 chown -R fluent /home/fluent
 chown -R fluent /fluentd
 
-exec gosu fluent "$@"
+exec su-exec fluent "$@"

lib/fluent/plugin/in_kubernetes_metrics.rb

@@ -72,8 +72,6 @@ def configure(conf)
 
         if @use_rest_client
           raise Fluentd::ConfigError, 'node_name is required' if @node_name.nil? || @node_name.empty?
-        else
-          raise Fluentd::ConfigError, 'node_names array is required' if @node_names.nil? || @node_names.empty? || (@node_names.length <= 0)
         end
 
         parse_tag
@@ -84,6 +82,7 @@ def start
         super
 
         timer_execute :metric_scraper, @interval, &method(:scrape_metrics)
+        timer_execute :cadvisor_metric_scraper, @interval, &method(:scrape_cadvisor_metrics)
       end
 
       def close
@@ -193,9 +192,11 @@ def initialize_rest_client
 
         if env_host && env_port
           @kubelet_url = "http://#{env_host}:#{env_port}/stats/summary"
+          @cadvisor_url = "http://#{env_host}:#{env_port}/metrics/cadvisor"
         end
 
         log.info("Use URL #{@kubelet_url} for creating client to query kubelet summary api")
+        log.info("Use URL #{@cadvisor_url} for creating client to query cadvisor metrics api")
       end
 
       # This method is used to set the options for sending a request to the kubelet api
@@ -215,6 +216,16 @@ def summary_api(node)
           end
       end
 
+      def cadvisor_proxy_api(node)
+        @summary_api =
+          begin
+            @client.discover unless @client.discovered
+            @client.rest_client["/nodes/#{node}:#{@kubelet_port}/proxy/metrics/cadvisor"].tap do |endpoint|
+              log.info("Use URL #{endpoint.url} for scraping metrics")
+            end
+          end
+      end
+
       def parse_time(metric_time)
         Fluent::EventTime.from_time Time.iso8601(metric_time)
       end
@@ -378,6 +389,67 @@ def handle_response(response)
         log.error "Failed to scrape metrics, error=#{error.inspect}"
         log.error_backtrace
       end
+
+      def cadvisor_request_options
+        options = { method: 'get', url: @cadvisor_url }
+        options
+      end
+
+      def emit_cadvisor_metrics(metrics)
+        metrics = metrics.split("\n")
+        for metric in metrics
+          if metric.include? "container_name="
+            if metric.match(/^((?!container_name="").)*$/) && metric[0] != '#'
+              metric_str, metric_val =  metric.split(" ")
+              first_occur = metric_str.index('{')
+              metric_name = metric_str[0..first_occur-1]
+              pod_name = metric.match(/pod_name="\S*"/).to_s
+              pod_name = pod_name.split('"')[1]
+              image_name = metric.match(/image="\S*"/).to_s
+              image_name = image_name.split('"')[1]
+              namespace = metric.match(/namespace="\S*"/).to_s
+              namespace = namespace.split('"')[1]
+              metric_labels = {'pod_name' => pod_name, 'image' => image_name, 'namespace' => namespace, 'value' => metric_val}
+                if metric.match(/^((?!container_name="POD").)*$/)
+                  tag = 'pod'
+                  tag = generate_tag("#{tag}#{metric_name.gsub('_', '.')}")
+                  tag = tag.gsub('container', '')
+                else
+                  container_name = metric.match(/container_name="\S*"/).to_s
+                  container_name = container_name.split('"')[1]
+                  container_label = {'container_name' => container_name}
+                  metric_labels.merge(container_label)
+                  tag = generate_tag("#{metric_name.gsub('_', '.')}")
+                end
+              router.emit tag, @scraped_at_cadvisor, metric_labels
+            end
+          end
+        end
+      end
+
+      def scrape_cadvisor_metrics
+        if @use_rest_client
+          response = RestClient::Request.execute cadvisor_request_options
+          handle_cadvisor_response(response)
+        else
+          response = cadvisor_proxy_api(@node_name).get(@client.headers)
+          handle_cadvisor_response(response)
+        end
+      end
+
+      # This method is used to handle responses from the kubelet summary api
+      def handle_cadvisor_response(response)
+        # Checking response codes only for a successful GET request viz., 2XX codes
+        if (response.code < 300) && (response.code > 199)
+          @scraped_at_cadvisor = Time.now
+          emit_cadvisor_metrics response.body
+        else
+          log.error "Expected 2xx from cadvisor metrics API, but got #{response.code}. Response body = #{response.body}"
+        end
+      rescue StandardError => e
+        log.error "Failed to scrape metrics, error=#{$ERROR_INFO}, #{e.inspect}"
+        log.error_backtrace
+      end
     end
   end
 end

test/helper.rb

@@ -22,6 +22,7 @@ def k8s_host() "generics-aws-node-name" end
   def k8s_port() "10255" end
   def k8s_url(path='api') "https://#{k8s_host}:#{k8s_port}/#{path}" end
   def kubelet_summary_api_url() "http://generics-aws-node-name:10255/stats/summary" end
+  def kubelet_cadvisor_api_url() "http://generics-aws-node-name:10255/cadvisor/metrics" end
 
   def stub_k8s_requests
     ENV['KUBERNETES_SERVICE_HOST'] = k8s_host
@@ -61,6 +62,13 @@ def stub_kubelet_summary_api
     }.close
   end
 
+  def stub_metrics_cadvisor
+    open(File.expand_path('../metrics_cadvisor.json', __FILE__)).tap { |f|
+      stub_request(:get, "#{kubelet_cadvisor_api_url}")
+        .to_return(body: f.read())
+    }.close
+  end
+
   def get_parsed_string
     parsed_string = nil
     open(File.expand_path('../unit.json', __FILE__)).tap { |f|