Update the collector column families from Ascii to UTF8

Cassandra returns InvalidRequestException when collector
tries to add a column and if the column data contains
unicode characters.

The issue is seen when any resource is created or deleted and
the resource name has unicode characters.

This patch fixes the issue by changing the collector column
family validation types from AsciiType to UTF8Type.
The patch also takes care of upgrading the existing column families
if they exist in the db.

(cherry-picked from commit 358765e)

Change-Id: I54ae9e756f7f67ce684ef9e3ba557038bc8f8731
Closes-bug: #1433021

src/analytics/db_handler.cc

@@ -1190,6 +1190,7 @@ bool FlowDataIpv4ObjectWalker<T>::for_each(pugi::xml_node& node) {
                 break;
             }
         case GenDb::DbDataType::AsciiType:
+        case GenDb::DbDataType::UTF8Type:
             {
                 std::string val = node.child_value();
                 TXMLProtocol::unescapeXMLControlChars(val);

src/gendb/cdb_if.cc

@@ -543,7 +543,13 @@ CdbIf::CdbIfTypeMapDef CdbIf::CdbIfTypeMap =
                 DbEncodeDoubleComposite,
                 DbDecodeDoubleComposite,
                 DbEncodeDoubleNonComposite,
-                DbDecodeDoubleNonComposite));
+                DbDecodeDoubleNonComposite))
+        (GenDb::DbDataType::UTF8Type,
+            CdbIf::CdbIfTypeInfo("UTF8Type",
+                DbEncodeStringComposite,
+                DbDecodeStringComposite,
+                DbEncodeStringNonComposite,
+                DbDecodeStringNonComposite));
 
 class CdbIf::CleanupTask : public Task {
 public:
@@ -904,7 +910,7 @@ bool CdbIf::DbDataTypeVecToCompositeType(std::string& res,
         res = it->second.cassandra_type_;
         return true;
     } else {
-        res = "CompositeType(";
+        res = "org.apache.cassandra.db.marshal.CompositeType(";
         std::vector<GenDb::DbDataType::type>::const_iterator it = db_type.begin();
         CdbIfTypeMapDef::iterator jt;
 
@@ -914,7 +920,7 @@ bool CdbIf::DbDataTypeVecToCompositeType(std::string& res,
 
         it++;
         for (; it != db_type.end(); it++) {
-            res.append(", ");
+            res.append(",");
             if ((jt = CdbIfTypeMap.find(*it)) == CdbIfTypeMap.end())
                 return false;
             res.append(jt->second.cassandra_type_);
@@ -1128,7 +1134,8 @@ bool CdbIf::Db_AddColumnfamily(const GenDb::NewCf& cf) {
 
         CdbIfCfInfo *cfinfo;
         if (Db_GetColumnfamily(&cfinfo, cf.cfname_)) {
-            if ((*cfinfo->cfdef_.get()).gc_grace_seconds != 0) {
+            if ((*cfinfo->cfdef_.get()).gc_grace_seconds != 0 ||
+                DB_IsCfSchemaChanged(cfinfo->cfdef_.get(), &cf_def)) {
                 CDBIF_LOG(DEBUG, "CFName: " << cf.cfname_ << " ID: " << 
                     (*cfinfo->cfdef_.get()).id);
                 cf_def.__set_id((*cfinfo->cfdef_.get()).id);
@@ -1190,7 +1197,8 @@ bool CdbIf::Db_AddColumnfamily(const GenDb::NewCf& cf) {
 
         CdbIfCfInfo *cfinfo;
         if (Db_GetColumnfamily(&cfinfo, cf.cfname_)) {
-            if ((*cfinfo->cfdef_.get()).gc_grace_seconds != 0) {
+            if (((*cfinfo->cfdef_.get()).gc_grace_seconds != 0) ||
+                DB_IsCfSchemaChanged(cfinfo->cfdef_.get(), &cf_def)) {
                 CDBIF_LOG(DEBUG, "CFName: " << cf.cfname_ << " ID: " << 
                     (*cfinfo->cfdef_.get()).id);
                 cf_def.__set_id((*cfinfo->cfdef_.get()).id);
@@ -1223,6 +1231,47 @@ bool CdbIf::Db_AddColumnfamily(const GenDb::NewCf& cf) {
     return true;
 }
 
+bool CdbIf::DB_IsCfSchemaChanged(org::apache::cassandra::CfDef *cfdef,
+                                 org::apache::cassandra::CfDef *newcfdef) {
+    if (cfdef->key_validation_class != newcfdef->key_validation_class) {
+        return true;
+    }
+
+    if ((newcfdef->default_validation_class.size()) &&
+        (cfdef->default_validation_class != newcfdef->default_validation_class)) {
+        return true;
+    }
+
+    if (cfdef->comparator_type != newcfdef->comparator_type) {
+        return true;
+    }
+
+    if (cfdef->column_metadata.size() != newcfdef->column_metadata.size()) {
+        return true;
+    }
+
+    std::vector<ColumnDef>::iterator cfdef_it = cfdef->column_metadata.begin();
+    std::vector<ColumnDef>::iterator newcfdef_it;
+
+    bool schema_changed;
+    for(; cfdef_it != cfdef->column_metadata.end(); cfdef_it++) {
+        schema_changed = true;
+        for(newcfdef_it = newcfdef->column_metadata.begin();
+            newcfdef_it != newcfdef->column_metadata.end(); newcfdef_it++) {
+            if(cfdef_it->name == newcfdef_it->name &&
+                cfdef_it->validation_class == newcfdef_it->validation_class) {
+                schema_changed = false;
+                break;
+             }
+        }
+        if(schema_changed) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
 bool CdbIf::Db_AsyncAddColumn(CdbIfColList &cl) {
     GenDb::ColList *new_colp(cl.gendb_cl);
     if (new_colp == NULL) {

src/gendb/cdb_if.h

@@ -89,7 +89,7 @@ class CdbIf : public GenDb::GenDbIf {
             DbDecodeCompositeFunc decode_composite_fn,
             DbEncodeNonCompositeFunc encode_non_composite_fn,
             DbDecodeNonCompositeFunc decode_non_composite_fn) :
-            cassandra_type_(cassandra_type),
+            cassandra_type_("org.apache.cassandra.db.marshal." + cassandra_type),
             encode_composite_fn_(encode_composite_fn),
             decode_composite_fn_(decode_composite_fn),
             encode_non_composite_fn_(encode_non_composite_fn),
@@ -162,6 +162,8 @@ class CdbIf : public GenDb::GenDbIf {
     bool Db_AsyncAddColumn(CdbIfColList &cl);
     bool Db_AsyncAddColumnLocked(CdbIfColList &cl);
     void Db_BatchAddColumn(bool done);
+    bool DB_IsCfSchemaChanged(org::apache::cassandra::CfDef *cfdef,
+                              org::apache::cassandra::CfDef *newcfdef);
     // Read
     static const int kMaxQueryRows = 5000;
     // API to get range of column data for a range of rows 

src/gendb/gendb.sandesh

@@ -16,6 +16,7 @@ enum DbDataType {
     Unsigned64Type    = 7, // uint64_t
 
     DoubleType        = 8, // double
+    UTF8Type          = 9, //utf-8 string
 }
 
 struct DbTableInfo {

src/query_engine/select.cc

@@ -464,6 +464,7 @@ query_status_t SelectQuery::process_query() {
                             break;
                           }
                     case GenDb::DbDataType::AsciiType:
+                    case GenDb::DbDataType::UTF8Type:
                           {
                             try {
                                 elem_value = boost::get<std::string>(kt->second);