[astute naily] Rework progress calculation. Add Deployment progress.

[puppet-nailgun] Add syslog transfer from target to master node.

[astute] Add deploying progress calculate.

[astute naily] Fix some misspells in porgress calculation.

[astute naily] Remove update task progress while deploy or provision nodes.

[astute naily] Move deploy log pattern to orchestrator.

[naily] Use LogParser class while provisioning.

[astute naily] Wrap LogParser into begin...rescue.

astute/examples/logparser.rb

@@ -0,0 +1,38 @@
+#!/usr/bin/env ruby
+
+require 'rubygems'
+require 'astute'
+
+
+
+deploy_pattern_spec = {'type' => 'count-lines', 'separator' => 'lololo',
+	'endlog_patterns' => [{'pattern' => /Finished catalog run in [0-9]+\.[0-9]* seconds\n/, 'progress' => 1.0}],
+	'expected_line_number' => 500}
+path='/var/tmp/1'
+#n = Astute::LogParser.get_log_progress(path, pattern_spec)
+
+path = '/home/adanin/anaconda.log'
+pattern_spec = {'type' => 'pattern-list', 'chunk_size' =>  10000, # Size of block which reads for pattern searching.
+'pattern_list' => [
+  {'pattern' => 'Running kickstart %%pre script', 'progress' => 0.08},
+  {'pattern' => 'to step enablefilesystems', 'progress' => 0.09},
+  {'pattern' => 'to step reposetup', 'progress' => 0.13},
+  {'pattern' => 'to step installpackages', 'progress' => 0.16},
+  {'pattern' => 'Installing',
+    'number' => 210, # Now it install 205 packets. Add 5 packets for growth in future.
+    'p_min' => 0.16, # min percent
+    'p_max' => 0.87 # max percent
+    },
+  # {'pattern' => 'to step postinstallconfig', 'progress' => 0.87},
+  # {'pattern' => 'to step dopostaction', 'progress' => 0.92},
+  ].reverse
+}
+#Astute::LogParser.add_log_separator(path)
+#n = Astute::LogParser.get_log_progress(path, pattern_spec)
+
+deployLogParser = Astute::LogParser::ParseNodeLogs.new('puppet-agent.log', deploy_pattern_spec)
+nodes = [{'uid' => '1', 'ip' => '1.1.1.1'}]
+n = deployLogParser.progress_calculate(nodes)
+#deployLogParser.add_separator(nodes)
+
+p n

astute/lib/astute.rb

@@ -1,6 +1,7 @@
 require 'json'
 require 'logger'
 
+require 'astute/logparser'
 require 'astute/orchestrator'
 require 'astute/metadata'
 require 'astute/deployment_engine'

astute/lib/astute/deployment_engine.rb

@@ -7,13 +7,17 @@ module Astute
   class DeploymentEngine
     def initialize(context)
       @ctx = context
+      @pattern_spec = {'type' => 'count-lines',
+        'endlog_patterns' => [{'pattern' => /Finished catalog run in [0-9]+\.[0-9]* seconds\n/, 'progress' => 1.0}],
+        'expected_line_number' => 500}
+      @deployLogParser = Astute::LogParser::ParseNodeLogs.new('puppet-agent.log', @pattern_spec)
     end
 
-    def deploy(nodes, attrs, prev_progress)
+    def deploy(nodes, attrs)
       # FIXME(mihgen): hardcode simple_compute for now
       attrs['deployment_mode'] ||= 'simple_compute'
       Astute.logger.info "Deployment mode #{attrs['deployment_mode']}, using #{self.class} for deployment."
-      result = self.send("deploy_#{attrs['deployment_mode']}", nodes, attrs, prev_progress)
+      result = self.send("deploy_#{attrs['deployment_mode']}", nodes, attrs)
     end
 
     def method_missing(method, *args)
@@ -23,22 +27,18 @@ module Astute
     # This method is called by Ruby metaprogramming magic from deploy method
     # It should not contain any magic with attributes, and should not directly run any type of MC plugins
     # It does only support of deployment sequence. See deploy_piece implementation in subclasses.
-    def deploy_simple_compute(nodes, attrs, prev_progress)
+    def deploy_simple_compute(nodes, attrs)
-      deploying_progress_part = 1 - prev_progress
-
       ctrl_nodes = nodes.select {|n| n['role'] == 'controller'}
 
       attrs = extend_attrs(nodes, attrs)
 
       deploy_piece(ctrl_nodes, attrs)
-      progress = (100* prev_progress + 40 * deploying_progress_part).to_i
-      @ctx.reporter.report({'progress' => progress})
 
+      @deployLogParser.pattern_spec['expected_line_number'] = 380
       compute_nodes = nodes.select {|n| n['role'] == 'compute'}
       deploy_piece(compute_nodes, attrs)
-      progress = (100* prev_progress + 60 * deploying_progress_part).to_i
-      @ctx.reporter.report({'progress' => progress})
 
+      @deployLogParser.pattern_spec['expected_line_number'] = 300
       other_nodes = nodes - ctrl_nodes - compute_nodes
       deploy_piece(other_nodes, attrs)
       return

astute/lib/astute/deployment_engine/nailyfact.rb

@@ -48,7 +48,7 @@ class Astute::DeploymentEngine::NailyFact < Astute::DeploymentEngine
     end
     Astute.logger.info "#{@ctx.task_id}: All required attrs/metadata passed via facts extension. Starting deployment."
 
-    Astute::PuppetdDeployer.deploy(@ctx, nodes)
+    Astute::PuppetdDeployer.deploy(@ctx, nodes, @deployLogParser)
     @ctx.reporter.report nodes_status(nodes, 'ready')
     nodes_roles = nodes.map { |n| { n['uid'] => n['role'] } }
     Astute.logger.info "#{@ctx.task_id}: Finished deployment of nodes => roles: #{nodes_roles.inspect}"

astute/lib/astute/logparser.rb

@@ -0,0 +1,176 @@
+module Astute
+  module LogParser
+    @separator = "SEPARATOR\n"
+    @log_portion = 10000
+
+    class ParseNodeLogs
+      attr_accessor :pattern_spec
+
+      def initialize(filename, pattern_spec)
+        @filename = filename
+        @pattern_spec = pattern_spec
+      end
+
+      def progress_calculate(nodes)
+        return 0 if nodes.empty?
+        nodes_progress = []
+        nodes.each do |node|
+          path = "/var/log/remote/#{node['ip']}/#{@filename}"
+          nodes_progress << {
+            'uid' => node['uid'],
+            'progress' => (LogParser::get_log_progress(path, @pattern_spec)*100).to_i # Return percent of progress
+          }
+        end
+        return nodes_progress
+      end
+
+      def add_separator(nodes)
+        nodes.each do |node|
+          path = "/var/log/remote/#{node['ip']}/#{@filename}"
+          LogParser::add_log_separator(path)
+        end
+      end
+    end
+
+
+    public
+    def self.add_log_separator(path, separator=@separator)
+      File.open(path, 'a') {|fo| fo.write separator } if File.readable?(path)
+    end
+
+    def self.get_log_progress(path, pattern_spec)
+      # Pattern specification example:
+      # pattern_spec = {'type' => 'pattern-list', 'separator' => "custom separator\n",
+      #   'chunk_size' => 10000,
+      #   'pattern_list' => [
+      #     {'pattern' => 'to step installpackages', 'progress' => 0.16},
+      #     {'pattern' => 'Installing',
+      #       'number' => 210, # Now it install 205 packets. Add 5 packets for growth in future.
+      #       'p_min' => 0.16, # min percent
+      #       'p_max' => 0.87 # max percent
+      #       }
+      #     ]
+      #   }
+
+      return 0 unless File.readable?(path)
+      progress = nil
+      File.open(path) do |fo|
+        # Try to find well-known ends of log.
+        endlog = find_endlog_patterns(fo, pattern_spec)
+        return endlog if endlog
+        # Start reading from end of file.
+        fo.pos = fo.stat.size
+
+        if pattern_spec['type'] == 'count-lines'
+          progress = simple_line_counter(fo, pattern_spec)
+        elsif pattern_spec['type'] = 'pattern-list'
+          progress = simple_pattern_finder(fo, pattern_spec)
+        end
+      end
+      unless progress
+        Naily.logger.warn("Wrong pattern #{pattern_spec.inspect} defined for calculating progress via logs.")
+        return 0
+      end
+      return progress
+    end
+
+    private
+    def self.simple_pattern_finder(fo, pattern_spec)
+      # Use custom separator if defined.
+      separator = pattern_spec['separator']
+      separator = @separator unless separator
+      log_patterns = pattern_spec['pattern_list']
+      unless log_patterns
+        Naily.logger.warn("Wrong pattern #{pattern_spec.inspect} defined for calculating progress via logs.")
+        return 0
+      end
+
+      chunk = get_chunk(fo, pattern_spec['chunk_size'])
+      pos = chunk.rindex(separator)
+      chunk = chunk.slice((pos + separator.size)..-1) if pos
+      block = chunk.split("\n")
+      return 0 unless block
+      while true
+        string = block.pop
+        return 0 unless string # If we found nothing
+        log_patterns.each do |pattern|
+          if string.include?(pattern['pattern'])
+            return pattern['progress'] if pattern['progress']
+            if pattern['number']
+              string = block.pop
+              counter = 1
+              while string
+                counter += 1 if string.include?(pattern['pattern'])
+                string = block.pop
+              end
+              progress = counter.to_f / pattern['number']
+              progress = 1 if progress > 1
+              progress = pattern['p_min'] + progress * (pattern['p_max'] - pattern['p_min'])
+              return progress
+            end
+            Naily.logger.warn("Wrong pattern #{pattern_spec.inspect} defined for calculating progress via log.")
+          end
+        end
+      end
+    end
+
+    def self.find_endlog_patterns(fo, pattern_spec)
+      endlog_patterns = pattern_spec['endlog_patterns']
+      return nil unless endlog_patterns
+      fo.pos = fo.stat.size
+      chunk = get_chunk(fo, 100)
+      endlog_patterns.each do |pattern|
+        return pattern['progress'] if chunk.end_with?(pattern['pattern'])
+      end
+      return nil
+    end
+
+    def self.simple_line_counter(fo, pattern_spec)
+      # Use custom separator if defined.
+      separator = pattern_spec['separator']
+      separator = @separator unless separator
+      counter = 0
+      end_of_scope = false
+      previous_subchunk = ''
+      until end_of_scope
+        chunk = get_chunk(fo, pattern_spec['chunk_size'])
+        break unless chunk
+        # Trying to find separator on border between chunks.
+        subchunk = chunk.slice((1-separator.size)..-1)
+        # End of file reached. Exit from cycle.
+        end_of_scope = true unless subchunk
+        if subchunk and (subchunk + previous_subchunk).include?(separator)
+          # Separator found on border between chunks. Exit from cycle.
+          end_of_scope = true
+          continue
+        end
+
+        pos = chunk.rindex(separator)
+        if pos
+          end_of_scope = true
+          chunk = chunk.slice((pos + separator.size)..-1)
+        end
+        counter += chunk.count("\n")
+      end
+      number = pattern_spec['expected_line_number']
+      unless number
+        Naily.logger.warn("Wrong pattern #{pattern_spec.inspect} defined for calculating progress via log.")
+        return 0
+      end
+      progress = counter.to_f / number
+      progress = 1 if progress > 1
+      return progress
+    end
+
+    def self.get_chunk(fo, size=nil)
+      size = @log_portion unless size
+      return nil if fo.pos == 0
+      size = fo.pos if fo.pos < size
+      next_pos = fo.pos - size
+      fo.pos = next_pos
+      block = fo.read(size)
+      fo.pos = next_pos
+      return block
+    end
+  end
+end

astute/lib/astute/orchestrator.rb

@@ -13,10 +13,10 @@ module Astute
       return systems.map {|n| {'uid' => n.results[:sender], 'node_type' => n.results[:data][:node_type].chomp}}
     end
 
-    def deploy(reporter, task_id, nodes, attrs, prev_progress=0)
+    def deploy(reporter, task_id, nodes, attrs)
       context = Context.new(task_id, reporter)
       deploy_engine_instance = @deploy_engine.new(context)
-      deploy_engine_instance.deploy(nodes, attrs, prev_progress)
+      deploy_engine_instance.deploy(nodes, attrs)
     end
 
     def remove_nodes(reporter, task_id, nodes)

astute/lib/astute/puppetd.rb

@@ -4,13 +4,14 @@ require 'timeout'
 module Astute
   module PuppetdDeployer
     private
+
     def self.calc_error_nodes(last_run_summary)
       error_nodes = last_run_summary.select { |n|
             n.results[:data][:resources]['failed'] != 0}.map {|x| x.results[:sender]}
       return error_nodes
     end
 
-    def self.wait_until_puppet_done(ctx, puppetd, prev_summary)
+    def self.wait_until_puppet_done(ctx, puppetd, prev_summary, nodes, deployLogParser)
       prev_error_nodes = calc_error_nodes(prev_summary)
       last_run = prev_summary
       while last_run.all? {|x| x.results[:data][:time]['last_run'] ==
@@ -28,15 +29,26 @@ module Astute
           # We want to send error message only once (only when status is changed)
           prev_error_nodes = new_error_nodes
         end
+        begin
+          nodes_progress = deployLogParser.progress_calculate(nodes)
+          ctx.reporter.report({'nodes' => nodes_progress})
+        rescue Exception => e
+          Astute.logger.warn "Some error occured when parse logs for nodes progress: #{e.message}, trace: #{e.backtrace.inspect}"
+        end
       end
       last_run
     end
 
     public
-    def self.deploy(ctx, nodes)
+    def self.deploy(ctx, nodes, deployLogParser)
       uids = nodes.map {|n| n['uid']}
       puppetd = MClient.new(ctx, "puppetd", uids)
       lastrun_summary = puppetd.last_run_summary
+      begin
+        deployLogParser.add_separator(nodes)
+      rescue Exception => e
+        Astute.logger.warn "Some error occured when add separator to logs: #{e.message}, trace: #{e.backtrace.inspect}"
+      end
 
       puppetd.runonce
 
@@ -46,7 +58,7 @@ module Astute
         # Yes, we polling here and yes, it's temporary.
         # As a better implementation we can later use separate queue to get result, ex. http://www.devco.net/archives/2012/08/19/mcollective-async-result-handling.php
         # or we can rewrite puppet agent not to fork, and increase ttl for mcollective RPC.
-        wait_until_puppet_done(ctx, puppetd, lastrun_summary)
+        wait_until_puppet_done(ctx, puppetd, lastrun_summary, nodes, deployLogParser)
       end
       summary = puppetd.last_run_summary.map { |x| {x.results[:sender] => x.results[:data][:resources]} }
       Astute.logger.debug "Results of puppet run from all nodes: #{summary.inspect}"

mcagent/puppetd.rb

@@ -131,7 +131,7 @@ module MCollective
       end
 
       def runonce_background
-        cmd = [@puppetd, "--onetime"]
+        cmd = [@puppetd, "--onetime", "--logdest", 'syslog']
 
         unless request[:forcerun]
           if @splaytime && @splaytime > 0

naily/lib/naily/dispatcher.rb

@@ -6,22 +6,22 @@ module Naily
       @orchestrator = Astute::Orchestrator.new
       @producer = producer
       @default_result = {'status' => 'ready', 'progress' => 100}
-      @provision_progress_part = 0.8
+      @pattern_spec = {'type' => 'pattern-list', 'chunk_size' =>  40000, # Size of block which reads for pattern searching.
-      @anaconda_log_portion = 40000 # Size of block which reads for pattern searching.
+        'pattern_list' => [
-      @ananconda_log_patterns = [
+          {'pattern' => 'Running kickstart %%pre script', 'progress' => 0.08},
-        {'pattern' => 'Running kickstart %%pre script', 'progress' => 0.08},
+          {'pattern' => 'to step enablefilesystems', 'progress' => 0.09},
-        {'pattern' => 'to step enablefilesystems', 'progress' => 0.09},
+          {'pattern' => 'to step reposetup', 'progress' => 0.13},
-        {'pattern' => 'to step reposetup', 'progress' => 0.13},
+          {'pattern' => 'to step installpackages', 'progress' => 0.16},
-        {'pattern' => 'to step installpackages', 'progress' => 0.16},
+          {'pattern' => 'Installing',
-        {'pattern' => 'Installing',
+            'number' => 210, # Now it install 205 packets. Add 5 packets for growth in future.
-          'number' => 210, # Now it install 205 packets. Add 5 packets for growth in future.
+            'p_min' => 0.16, # min percent
-          'p_min' => 0.16, # min percent
+            'p_max' => 0.87 # max percent
-          'p_max' => 0.87 # max percent
+            },
-          },
+          {'pattern' => 'to step postinstallconfig', 'progress' => 0.87},
-        {'pattern' => 'to step postinstallconfig', 'progress' => 0.87},
+          {'pattern' => 'to step dopostaction', 'progress' => 0.92},
-        {'pattern' => 'to step dopostaction', 'progress' => 0.92},
+          ].reverse
-        {'pattern' => 'SEPARATOR', 'progress' => 0}
+        }
-        ].reverse
+      @provisionLogParser = Astute::LogParser::ParseNodeLogs.new('install/anaconda.log', @pattern_spec)
     end
 
     def echo(args)
@@ -35,7 +35,11 @@ module Naily
       time = Time::now.to_f
       nodes_not_booted = nodes.map { |n| n['uid'] }
       Naily.logger.info "Starting OS provisioning for nodes: #{nodes_not_booted.join(',')}" 
-      add_anaconda_log_separator(nodes)
+      begin
+        @provisionLogParser.add_separator(nodes)
+      rescue Exception => e
+        Naily.logger.warn "Some error occured when add separator to logs: #{e.message}, trace: #{e.backtrace.inspect}"
+      end
       time = 10 + time - Time::now.to_f
       sleep (time) if time > 0 # Wait while nodes going to reboot. Sleep not greater than 10 sec.
       begin
@@ -50,21 +54,19 @@ module Naily
               break
             end
             nodes_not_booted = nodes.map { |n| n['uid'] } - types.map { |n| n['uid'] }
-            all_progress = 0
+            begin
-            nodes_progress = provisioning_progress_calculate(nodes)
+              nodes_progress = @provisionLogParser.progress_calculate(nodes)
-            nodes_progress.each do |n|
+              nodes_progress.each do |n|
-              if target_uids.include?(n['uid'])
+                if target_uids.include?(n['uid'])
-                all_progress += 100 # 100%
+                  n['progress'] = 100
-                n['progress'] = 100
+                  # TODO(mihgen): should we change status only once?
-                # TODO(mihgen): should we change status only once?
+                  n['status'] = 'provisioned'
-                n['status'] = 'provisioned'
+                end
-              else
-                all_progress += n['progress']
               end
+              reporter.report({'nodes' => nodes_progress})
+            rescue Exception => e
+              Naily.logger.warn "Some error occured when parse logs for nodes progress: #{e.message}, trace: #{e.backtrace.inspect}"
             end
-
-            all_progress = (all_progress * @provision_progress_part / nodes.size).to_i
-            reporter.report({'progress' => all_progress, 'nodes' => nodes_progress})
             time = 5 + time - Time::now.to_f
             sleep (time) if time > 0 # Sleep not greater than 5 sec.
           end
@@ -81,8 +83,12 @@ module Naily
         return
       end
 
-      reporter.report({'progress' => (@provision_progress_part * 100).to_i})
+      nodes_progress = nodes.map do |n|
-      result = @orchestrator.deploy(reporter, data['args']['task_uuid'], nodes, data['args']['attributes'], @provision_progress_part)
+        {'uid' => n['uid'], 'progress' => 100, 'status' => 'provisioned'}
+      end
+      reporter.report({'nodes' => nodes_progress})
+
+      result = @orchestrator.deploy(reporter, data['args']['task_uuid'], nodes, data['args']['attributes'])
       report_result(result, reporter)
     end
 
@@ -101,61 +107,6 @@ module Naily
     end
 
     private
-    def provisioning_progress_calculate(nodes)
-      return 0 if nodes.empty?
-      nodes_progress = []
-      nodes.each do |node|
-        path = "/var/log/remote/#{node['ip']}/install/anaconda.log"
-        nodes_progress << {
-          'uid' => node['uid'],
-          'progress' => (get_anaconda_log_progress(path)*100).to_i # Return percent of progress
-        }
-      end
-      return nodes_progress
-    end
-
-    def add_anaconda_log_separator(nodes)
-      nodes.each do |node|
-        path = "/var/log/remote/#{node['ip']}/install/anaconda.log"
-        File.open(path, 'a') {|fo| fo.puts "\n\nSEPARATOR" } if File.readable?(path)
-      end
-    end
-
-    def get_anaconda_log_progress(file)
-      return 0 unless File.readable?(file)
-      File.open(file) do |fo|
-        if fo.stat.size > @anaconda_log_portion
-          fo.pos = fo.stat.size - @anaconda_log_portion
-          block = fo.read(@anaconda_log_portion).split("\n")
-        else
-          block = fo.read(fo.stat.size).split("\n")
-        end
-
-        while true
-          string = block.pop
-          return 0 unless string # If we found nothing
-          @ananconda_log_patterns.each do |pattern|
-            if string.include?(pattern['pattern'])
-              return pattern['progress'] if pattern['progress']
-              if pattern['number']
-                string = block.pop
-                counter = 1
-                while string
-                  counter += 1 if string.include?(pattern['pattern'])
-                  string = block.pop
-                end
-                progress = counter.to_f / pattern['number']
-                progress = 1 if progress > 1
-                progress = pattern['p_min'] + progress * (pattern['p_max'] - pattern['p_min'])
-                return progress
-              end
-              Naily.logger.warn("Wrong pattern #{pattern.inspect} defined for calculating progress via Anaconda log.")
-            end
-          end
-        end
-      end
-    end
-
     def report_result(result, reporter)
       result = {} unless result.instance_of?(Hash)
       status = @default_result.merge(result)

puppet/nailgun/manifests/cobbler.pp

@@ -56,6 +56,14 @@ class nailgun::cobbler(
     require => Class["cobbler::server"],
   }
 
+  file {"/var/lib/cobbler/snippets/target_logs_to_master":
+    content => template("nailgun/cobbler/target_logs_to_master.snippet.erb"),
+    owner => root,
+    group => root,
+    mode => 0644,
+    require => Class["cobbler::server"],
+  }
+
   file {"/var/lib/cobbler/snippets/kickstart_ntp":
     content => template("nailgun/cobbler/kickstart_ntp.snippet.erb"),
     owner => root,

puppet/nailgun/templates/cobbler/centos.ks.erb

@@ -188,7 +188,10 @@ $SNIPPET('ssh_disable_gssapi')
 $SNIPPET('ssh_disable_dns')
 
 # Copying default bash settings to the root directory
-cp -f /etc/skel/.bash* /root/ 
+cp -f /etc/skel/.bash* /root/
+
+# Rsyslogd should send all messages to master node
+$SNIPPET('target_logs_to_master')
 
 # COBBLER EMBEDDED SNIPPET: 'kickstart_done'
 # DISABLES PXE BOOTING

puppet/nailgun/templates/cobbler/target_logs_to_master.snippet.erb

@@ -0,0 +1,7 @@
+#raw
+cat >/etc/rsyslog.d/10-log2master.conf <<EOF
+# Log all messages to master node
+\$template LogToMaster, "<%%PRI%>1 %\$NOW%T%TIMESTAMP:8:\$%Z %HOSTNAME% %APP-NAME% %PROCID% %MSGID% -%msg%\n"
+#end raw
+*.* @$server;LogToMaster
+EOF

puppet/nailgun/templates/rsyslog/30-remote-log.conf.erb

@@ -3,7 +3,7 @@ $ModLoad imudp
 $UDPServerRun 514
 
 # remote anaconda logs
-$template RemoteLog, "%FROMHOST% %TIMESTAMP% %HOSTNAME% %APP-NAME%: %msg%\n"
+$template RemoteLog, "%FROMHOST% %TIMESTAMP% %HOSTNAME% %APP-NAME%: %syslogseverity-text%: %msg%\n"
 $template RemoteLogFile, "/var/log/remote/%FROMHOST%/%APP-NAME%.log"
 :FROMHOST, regex, "^[1-9]" ?RemoteLogFile;RemoteLog
 :FROMHOST, regex, "^[1-9]" ~