
/*
 * Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package monasca.persister.repository;

import com.google.inject.Inject;

import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormatter;
import org.joda.time.format.ISODateTimeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import io.dropwizard.setup.Environment;
import monasca.persister.repository.influxdb.InfluxPoint;

public class InfluxV9MetricRepo extends InfluxMetricRepo {

  private static final Logger logger = LoggerFactory.getLogger(InfluxV9MetricRepo.class);

  private final SimpleDateFormat simpleDateFormat =
      new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS zzz");

  private final InfluxV9RepoWriter influxV9RepoWriter;

  @Inject
  public InfluxV9MetricRepo(final Environment env,
                            final InfluxV9RepoWriter influxV9RepoWriter) {

    super(env);
    this.influxV9RepoWriter = influxV9RepoWriter;

  }

  @Override
  protected void write() throws Exception {

    this.influxV9RepoWriter.write(getInfluxPointArry());

  }

  private InfluxPoint[] getInfluxPointArry() throws Exception {

    DateTimeFormatter dateFormatter = ISODateTimeFormat.dateTime();

    List<InfluxPoint> influxPointList = new LinkedList<>();

    for (final Sha1HashId defDimId : this.measurementMap.keySet()) {

      final DefDim defDim = this.defDimMap.get(defDimId);
      final Def def = getDef(defDim.defId);
      final Set<Dim> dimSet = getDimSet(defDim.dimId);

      Map<String, String> tagMap = new HashMap<>();
      for (Dim dim : dimSet) {
        tagMap.put(dim.name, dim.value);
      }
      tagMap.put("tenant_id", def.tenantId);
      tagMap.put("region", def.region);

      for (final Measurement measurement : this.measurementMap.get(defDimId)) {

        Date date = this.simpleDateFormat.parse(measurement.time + " UTC");
        DateTime dateTime = new DateTime(date.getTime(), DateTimeZone.UTC);
        String dateString = dateFormatter.print(dateTime);

        Map<String, Object> valueMap = new HashMap<>();
        valueMap.put("value", measurement.value);

        InfluxPoint influxPoint = new InfluxPoint(def.name, tagMap, dateString, valueMap);

        influxPointList.add(influxPoint);

        this.measurementMeter.mark();
      }
    }

    return influxPointList.toArray(new InfluxPoint[influxPointList.size()]);
  }
}

@@ -661,24 +661,136 @@
 epository;%0A%0A
+import com.fasterxml.jackson.core.JsonProcessingException;%0Aimport com.fasterxml.jackson.databind.ObjectMapper;%0A%0A
 import com.g
@@ -1456,24 +1456,89 @@
 po.class);%0A%0A
+  private final ObjectMapper objectMapper = new ObjectMapper();%0A%0A
   private fi
@@ -3058,24 +3058,550 @@
 ent.value);%0A
+        String valueMetaJson = null;%0A        if (measurement.valueMeta != null && !measurement.valueMeta.isEmpty()) %7B%0A          try %7B%0A            valueMetaJson = objectMapper.writeValueAsString(measurement.valueMeta);%0A            logger.debug(%22Added value for value_meta of %7B%7D%22, valueMetaJson);%0A          %7D catch (JsonProcessingException e) %7B%0A            logger.error(%22Unable to serialize %22 + measurement.valueMeta, e);%0A            valueMetaJson = null;%0A          %7D%0A        %7D%0A        valueMap.put(%22value_meta%22, valueMetaJson);
 %0A        Inf

Add valueMeta support for Influx DB V9