HPCC-16011 Add a daliadmin dfscheck command to locate dfs problems

Signed-off-by: Gavin Halliday <gavin.halliday@lexisnexis.com>

dali/daliadmin/daliadmin.cpp

@@ -80,6 +80,7 @@ void usage(const char *exe)
   printf("Logical File meta information commands:\n");
   printf("  dfsfile <logicalname>          -- get meta information for file\n");
   printf("  dfspart <logicalname> <part>   -- get meta information for part num\n");
+  printf("  dfscheck                       -- verify dfs file information is valid\n");
   printf("  dfscsv <logicalnamemask>       -- get csv info. for files matching mask\n");
   printf("  dfsgroup <logicalgroupname> [filename] -- get IPs for logical group (aka cluster). Written to optional filename if provided\n");
   printf("  clusternodes <clustername> [filename] -- get IPs for cluster group. Written to optional filename if provided\n");
@@ -658,6 +659,43 @@ static void writeGroup(IGroup *group, const char *name, const char *outputFilena
     }
 }
 
+unsigned dfsCheck(StringBuffer & path, IPropertyTree * tree)
+{
+    const char * name = tree->queryProp("@name");
+    //MORE: What other consistency checks can be added here?
+    if (tree->hasProp("Attr[2]"))
+    {
+        printf("%s%s - duplicate Attr tag\n", path.str(), name ? name : "");
+        return 1;
+    }
+
+    unsigned issues = 0;
+    unsigned prevLength = path.length();
+    if (name)
+        path.append(name).append("::");
+    Owned<IPropertyTreeIterator> elems = tree->getElements("*");
+    ForEach(*elems)
+    {
+        issues += dfsCheck(path, &elems->query());
+    }
+    path.setLength(prevLength);
+    return issues;
+}
+
+void dfsCheck()
+{
+    StringBuffer xpath;
+    Owned<IRemoteConnection> conn = connectXPathOrFile("/Files",true,xpath);
+    if (!conn) {
+        ERRLOG("Could not connect to %s","/Files");
+        return;
+    }
+
+    StringBuffer path;
+    dfsCheck(path, conn->queryRoot());
+}
+
+
 static void dfsGroup(const char *name, const char *outputFilename)
 {
     Owned<IGroup> group = queryNamedGroupStore().lookup(name);
@@ -2881,6 +2919,10 @@ int main(int argc, char* argv[])
                         CHECKPARAMS(2,2);
                         dfspart(params.item(1),userDesc,atoi(params.item(2)));
                     }
+                    else if (stricmp(cmd,"dfscheck")==0) {
+                        CHECKPARAMS(0,0);
+                        dfsCheck();
+                    }
                     else if (stricmp(cmd,"dfscsv")==0) {
                         CHECKPARAMS(1,1);
                         dfscsv(params.item(1),userDesc);