<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.beam.sdk.testutils.publishing;

import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.isNull;
import static java.util.Objects.requireNonNull;
import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNoneBlank;

import com.google.auto.value.AutoValue;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.apache.beam.sdk.testutils.NamedTestResult;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Collections2;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableSet;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.apache.commons.compress.utils.Charsets;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.entity.GzipCompressingEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.checkerframework.dataflow.qual.Pure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class InfluxDBPublisher {
  private static final Logger LOG = LoggerFactory.getLogger(InfluxDBPublisher.class);

  private InfluxDBPublisher() {}

  /** InfluxDB data point. */
  @AutoValue
  public abstract static class DataPoint {
    DataPoint() {}

    public abstract @Pure String measurement();

    public abstract @Pure Map<String, String> tags();

    public abstract @Pure Map<String, Number> fields();

    @Nullable
    public abstract @Pure Long timestamp();

    public abstract @Pure TimeUnit timestampUnit();

    @Override
    public final String toString() {
      return append(new StringBuilder()).toString();
    }

    private @Nullable Long timestampSecs() {
      return timestamp() != null ? timestampUnit().toSeconds(timestamp()) : null;
    }

    private StringBuilder append(StringBuilder builder) {
      return addMeasurement(builder, measurement(), tags(), fields(), timestampSecs());
    }
  }

  /** Creates an InfluxDB data point using optional custom Unix timestamp in seconds. */
  public static DataPoint dataPoint(
      String measurement,
      Map<String, String> tags,
      Map<String, Number> fields,
      @Nullable Long timestampSecs) {
    return new AutoValue_InfluxDBPublisher_DataPoint(
        measurement, tags, fields, timestampSecs, TimeUnit.SECONDS);
  }

  /** Creates an InfluxDB data point. */
  public static DataPoint dataPoint(
      String measurement,
      Map<String, String> tags,
      Map<String, Number> fields,
      @Nullable Long timestamp,
      TimeUnit timestampUnit) {
    return new AutoValue_InfluxDBPublisher_DataPoint(
        measurement, tags, fields, timestamp, timestampUnit);
  }

  /** @deprecated Use {@link #publish} instead. */
  @Deprecated
  public static void publishNexmarkResults(
      final Collection<Map<String, Object>> results,
      final InfluxDBSettings settings,
      final Map<String, String> tags) {
    publishWithCheck(settings, nexmarkDataPoints(results, tags));
  }

  public static void publishWithSettings(
      final Collection<NamedTestResult> results, final InfluxDBSettings settings) {
    @SuppressWarnings("nullness")
    Collection<DataPoint> dataPoints =
        Collections2.transform(results, res -> res.toInfluxDBDataPoint(settings.measurement));
    publish(settings, dataPoints);
  }

  public static void publish(
      final InfluxDBSettings settings, final Collection<DataPoint> dataPoints) {
    final StringBuilder builder = new StringBuilder();
    dataPoints.forEach(m -> m.append(builder).append('\n'));
    publishWithCheck(settings, builder.toString());
  }

  private static void publishWithCheck(final InfluxDBSettings settings, final String data) {
    requireNonNull(settings, "InfluxDB settings must not be null");
    if (isNoneBlank(settings.measurement, settings.database)) {
      try {
        final HttpClientBuilder builder = provideHttpBuilder(settings);
        final HttpPost postRequest = providePOSTRequest(settings);
        postRequest.setEntity(new GzipCompressingEntity(new ByteArrayEntity(data.getBytes(UTF_8))));
        executeWithVerification(postRequest, builder);
      } catch (Exception exception) {
        LOG.warn("Unable to publish metrics due to error: {}", exception.getMessage());
      }
    } else {
      LOG.warn("Missing property -- measurement/database. Metrics won't be published.");
    }
  }

  /** @deprecated To be removed, kept for legacy interface {@link #publishNexmarkResults} */
  @VisibleForTesting
  @Deprecated
  static String nexmarkDataPoints(
      final Collection<Map<String, Object>> results, final Map<String, String> tags) {
    final StringBuilder builder = new StringBuilder();
    final Set<String> fields = ImmutableSet.of("runtimeMs", "numResults");
    results.forEach(
        map -> {
          String measurement = checkArgumentNotNull(map.get("measurement")).toString();
          addMeasurement(builder, measurement, tags, filterKeys(map, fields), map.get("timestamp"))
              .append('\n');
        });
    return builder.toString();
  }
<commit_msg>Fix validation of measurement name in InfluxDBPublisher (addresses #22238)<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.beam.sdk.testutils.publishing;

import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.isNull;
import static java.util.Objects.requireNonNull;
import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNoneBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;

import com.google.auto.value.AutoValue;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.apache.beam.sdk.testutils.NamedTestResult;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Collections2;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableSet;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.apache.commons.compress.utils.Charsets;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.entity.GzipCompressingEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.checkerframework.dataflow.qual.Pure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class InfluxDBPublisher {
  private static final Logger LOG = LoggerFactory.getLogger(InfluxDBPublisher.class);

  private InfluxDBPublisher() {}

  /** InfluxDB data point. */
  @AutoValue
  public abstract static class DataPoint {
    DataPoint() {}

    public abstract @Pure String measurement();

    public abstract @Pure Map<String, String> tags();

    public abstract @Pure Map<String, Number> fields();

    @Nullable
    public abstract @Pure Long timestamp();

    public abstract @Pure TimeUnit timestampUnit();

    @Override
    public final String toString() {
      return append(new StringBuilder()).toString();
    }

    private @Nullable Long timestampSecs() {
      return timestamp() != null ? timestampUnit().toSeconds(timestamp()) : null;
    }

    private StringBuilder append(StringBuilder builder) {
      return addMeasurement(builder, measurement(), tags(), fields(), timestampSecs());
    }
  }

  /** Creates an InfluxDB data point using optional custom Unix timestamp in seconds. */
  public static DataPoint dataPoint(
      String measurement,
      Map<String, String> tags,
      Map<String, Number> fields,
      @Nullable Long timestampSecs) {
    Preconditions.checkArgument(isNotBlank(measurement), "Measurement cannot be blank");
    return new AutoValue_InfluxDBPublisher_DataPoint(
        measurement, tags, fields, timestampSecs, TimeUnit.SECONDS);
  }

  /** Creates an InfluxDB data point. */
  public static DataPoint dataPoint(
      String measurement,
      Map<String, String> tags,
      Map<String, Number> fields,
      @Nullable Long timestamp,
      TimeUnit timestampUnit) {
    Preconditions.checkArgument(isNotBlank(measurement), "Measurement cannot be blank");
    return new AutoValue_InfluxDBPublisher_DataPoint(
        measurement, tags, fields, timestamp, timestampUnit);
  }

  /** @deprecated Use {@link #publish} instead. */
  @Deprecated
  public static void publishNexmarkResults(
      final Collection<Map<String, Object>> results,
      final InfluxDBSettings settings,
      final Map<String, String> tags) {
    publishWithCheck(settings, nexmarkDataPoints(results, tags));
  }

  public static void publishWithSettings(
      final Collection<NamedTestResult> results, final InfluxDBSettings settings) {
    if (isNotBlank(settings.measurement)) {
      @SuppressWarnings("nullness")
      Collection<DataPoint> dataPoints =
          Collections2.transform(results, res -> res.toInfluxDBDataPoint(settings.measurement));
      publish(settings, dataPoints);
    } else {
      LOG.warn("Missing setting InfluxDB measurement. Metrics won't be published.");
    }
  }

  public static void publish(
      final InfluxDBSettings settings, final Collection<DataPoint> dataPoints) {
    final StringBuilder builder = new StringBuilder();
    dataPoints.forEach(m -> m.append(builder).append('\n'));
    publishWithCheck(settings, builder.toString());
  }

  private static void publishWithCheck(final InfluxDBSettings settings, final String data) {
    requireNonNull(settings, "InfluxDB settings must not be null");
    if (isNotBlank(settings.database)) {
      try {
        final HttpClientBuilder builder = provideHttpBuilder(settings);
        final HttpPost postRequest = providePOSTRequest(settings);
        postRequest.setEntity(new GzipCompressingEntity(new ByteArrayEntity(data.getBytes(UTF_8))));
        executeWithVerification(postRequest, builder);
      } catch (Exception exception) {
        LOG.warn("Unable to publish metrics due to error: {}", exception.getMessage());
      }
    } else {
      LOG.warn("Missing setting InfluxDB database. Metrics won't be published.");
    }
  }

  /** @deprecated To be removed, kept for legacy interface {@link #publishNexmarkResults} */
  @VisibleForTesting
  @Deprecated
  static String nexmarkDataPoints(
      final Collection<Map<String, Object>> results, final Map<String, String> tags) {
    final StringBuilder builder = new StringBuilder();
    final Set<String> fields = ImmutableSet.of("runtimeMs", "numResults");
    results.forEach(
        map -> {
          String measurement = checkArgumentNotNull(map.get("measurement")).toString();
          addMeasurement(builder, measurement, tags, filterKeys(map, fields), map.get("timestamp"))
              .append('\n');
        });
    return builder.toString();
  }

{'fraction_non_alphanumeric': 0.09973951192761174, 'fraction_numerical': 0.004318618042226488, 'mean_word_length': 4.381409074142383, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7156345', 'n_tokens_mistral': 4084, 'n_tokens_neox': 3999, 'n_words': 1054}