                           pager-data)))))
       (-> (resp/response "Page not found")
           (resp/status 404)))))
+
+(defnk set-log-file-permissions [fname root-dir]
+  (let [file (.getCanonicalFile (File. root-dir fname))
+        run-as-user (*STORM-CONF* SUPERVISOR-RUN-WORKER-AS-USER)
+        parent (.getParent (File. root-dir fname))
+        md-file (if (nil? parent) nil (get-metadata-file-for-wroker-logdir parent))
+        topo-owner (if (nil? md-file) nil (get-topo-owner-from-metadata-file (.getCanonicalPath md-file)))]
+    (when (and run-as-user
+          (not-nil? topo-owner)
+          (.exists file)
+          (not (Files/isReadable (.toPath file))))     
+      (log-debug "Setting permissions on file " fname " with topo-owner " topo-owner)
+      (ClientSupervisorUtils/processLauncherAndWait *STORM-CONF* topo-owner ["blob" (.getCanonicalPath file)] nil (str "setup group read permissions for file: " fname)))))
 
 (defnk download-log-file [fname req resp user ^String root-dir :is-daemon false]
   (let [file (.getCanonicalFile (File. root-dir fname))]

      (.mark logviewer:num-download-log-file-http-requests)
      (let [user (.getUserName http-creds-handler servlet-request)
            file (URLDecoder/decode (:file m))]
        (download-log-file file servlet-request servlet-response user log-root))
      (catch InvalidRequestException ex
        (log-error ex)

  (:use [org.apache.storm config daemon-config util log])
  (:use [org.apache.storm.ui helpers])
  (:import [org.apache.storm StormTimer]
           [org.apache.storm.daemon.supervisor ClientSupervisorUtils]
           [org.apache.storm.daemon.supervisor SupervisorUtils]
           [org.apache.storm.metric StormMetricsRegistry])
  (:import [org.apache.storm.utils Time VersionInfo ConfigUtils Utils ServerUtils ServerConfigUtils])
