[ONOS-7683] Add default implementation of telemetry config managers
Change-Id: Iac3681ceff2b343746d2d0b1043300044043b63a
diff --git a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryConfigManager.java b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryConfigManager.java
index db319a3..242709d 100644
--- a/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryConfigManager.java
+++ b/apps/openstacktelemetry/app/src/main/java/org/onosproject/openstacktelemetry/impl/KafkaTelemetryConfigManager.java
@@ -15,16 +15,197 @@
*/
package org.onosproject.openstacktelemetry.impl;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Modified;
+import org.apache.felix.scr.annotations.Property;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.onlab.util.Tools;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.openstacktelemetry.api.KafkaTelemetryAdminService;
import org.onosproject.openstacktelemetry.api.KafkaTelemetryConfigService;
import org.onosproject.openstacktelemetry.api.config.TelemetryConfig;
+import org.onosproject.openstacktelemetry.config.DefaultKafkaTelemetryConfig;
+import org.osgi.service.component.ComponentContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Dictionary;
+
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_BATCH_SIZE;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_KEY_SERIALIZER;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_LINGER_MS;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_MEMORY_BUFFER;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_REQUIRED_ACKS;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_RETRIES;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_SERVER_IP;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_SERVER_PORT;
+import static org.onosproject.openstacktelemetry.api.Constants.DEFAULT_KAFKA_VALUE_SERIALIZER;
/**
* Kafka server configuration manager for publishing openstack telemetry.
*/
public class KafkaTelemetryConfigManager implements KafkaTelemetryConfigService {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private static final String ADDRESS = "address";
+ private static final String PORT = "port";
+ private static final String RETRIES = "retries";
+ private static final String REQUIRED_ACKS = "requiredAcks";
+ private static final String BATCH_SIZE = "batchSize";
+ private static final String LINGER_MS = "lingerMs";
+ private static final String MEMORY_BUFFER = "memoryBuffer";
+ private static final String KEY_SERIALIZER = "keySerializer";
+ private static final String VALUE_SERIALIZER = "valueSerializer";
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ComponentConfigService componentConfigService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected KafkaTelemetryAdminService kafkaTelemetryAdminService;
+
+ @Property(name = ADDRESS, value = DEFAULT_KAFKA_SERVER_IP,
+ label = "Default IP address to establish initial connection to Kafka server")
+ protected String address = DEFAULT_KAFKA_SERVER_IP;
+
+ @Property(name = PORT, intValue = DEFAULT_KAFKA_SERVER_PORT,
+ label = "Default port number to establish initial connection to Kafka server")
+ protected Integer port = DEFAULT_KAFKA_SERVER_PORT;
+
+ @Property(name = RETRIES, intValue = DEFAULT_KAFKA_RETRIES,
+ label = "Number of times the producer can retry to send after first failure")
+ protected int retries = DEFAULT_KAFKA_RETRIES;
+
+ @Property(name = REQUIRED_ACKS, value = DEFAULT_KAFKA_REQUIRED_ACKS,
+ label = "Producer will get an acknowledgement after the leader has replicated the data")
+ protected String requiredAcks = DEFAULT_KAFKA_REQUIRED_ACKS;
+
+ @Property(name = BATCH_SIZE, intValue = DEFAULT_KAFKA_BATCH_SIZE,
+ label = "The largest record batch size allowed by Kafka")
+ protected Integer batchSize = DEFAULT_KAFKA_BATCH_SIZE;
+
+ @Property(name = LINGER_MS, intValue = DEFAULT_KAFKA_LINGER_MS,
+ label = "The producer groups together any records that arrive in " +
+ "between request transmissions into a single batched request")
+ protected Integer lingerMs = DEFAULT_KAFKA_LINGER_MS;
+
+ @Property(name = MEMORY_BUFFER, intValue = DEFAULT_KAFKA_MEMORY_BUFFER,
+ label = "The total memory used for log cleaner I/O buffers across all cleaner threads")
+ protected Integer memoryBuffer = DEFAULT_KAFKA_MEMORY_BUFFER;
+
+ @Property(name = KEY_SERIALIZER, value = DEFAULT_KAFKA_KEY_SERIALIZER,
+ label = "Serializer class for key that implements the Serializer interface")
+ protected String keySerializer = DEFAULT_KAFKA_KEY_SERIALIZER;
+
+ @Property(name = VALUE_SERIALIZER, value = DEFAULT_KAFKA_VALUE_SERIALIZER,
+ label = "Serializer class for value that implements the Serializer interface")
+ protected String valueSerializer = DEFAULT_KAFKA_VALUE_SERIALIZER;
+
+ @Activate
+ protected void activate(ComponentContext context) {
+ componentConfigService.registerProperties(getClass());
+ kafkaTelemetryAdminService.start(getConfig());
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ componentConfigService.unregisterProperties(getClass(), false);
+ kafkaTelemetryAdminService.stop();
+ log.info("Stopped");
+ }
+
+ @Modified
+ private void modified(ComponentContext context) {
+ readComponentConfiguration(context);
+ kafkaTelemetryAdminService.restart(getConfig());
+ log.info("Modified");
+ }
+
@Override
public TelemetryConfig getConfig() {
- return null;
+ return new DefaultKafkaTelemetryConfig.DefaultBuilder()
+ .withAddress(address)
+ .withPort(port)
+ .withRetries(retries)
+ .withRequiredAcks(requiredAcks)
+ .withBatchSize(batchSize)
+ .withLingerMs(lingerMs)
+ .withMemoryBuffer(memoryBuffer)
+ .withKeySerializer(keySerializer)
+ .withValueSerializer(valueSerializer)
+ .build();
+ }
+
+ /**
+ * Extracts properties from the component configuration context.
+ *
+ * @param context the component context
+ */
+ private void readComponentConfiguration(ComponentContext context) {
+ Dictionary<?, ?> properties = context.getProperties();
+
+ String addressStr = Tools.get(properties, ADDRESS);
+ address = addressStr != null ? addressStr : DEFAULT_KAFKA_SERVER_IP;
+ log.info("Configured. Kafka server address is {}", address);
+
+ Integer portConfigured = Tools.getIntegerProperty(properties, PORT);
+ if (portConfigured == null) {
+ port = DEFAULT_KAFKA_SERVER_PORT;
+ log.info("Kafka server port is NOT configured, default value is {}", port);
+ } else {
+ port = portConfigured;
+ log.info("Configured. Kafka server port is {}", port);
+ }
+
+ Integer retriesConfigured = Tools.getIntegerProperty(properties, RETRIES);
+ if (retriesConfigured == null) {
+ retries = DEFAULT_KAFKA_RETRIES;
+ log.info("Kafka number of retries property is NOT configured, default value is {}", retries);
+ } else {
+ retries = retriesConfigured;
+ log.info("Configured. Kafka number of retries is {}", retries);
+ }
+
+ String requiredAcksStr = Tools.get(properties, REQUIRED_ACKS);
+ requiredAcks = requiredAcksStr != null ? requiredAcksStr : DEFAULT_KAFKA_REQUIRED_ACKS;
+ log.info("Configured, Kafka required acknowledgement is {}", requiredAcks);
+
+ Integer batchSizeConfigured = Tools.getIntegerProperty(properties, BATCH_SIZE);
+ if (batchSizeConfigured == null) {
+ batchSize = DEFAULT_KAFKA_BATCH_SIZE;
+ log.info("Kafka batch size property is NOT configured, default value is {}", batchSize);
+ } else {
+ batchSize = batchSizeConfigured;
+ log.info("Configured. Kafka batch size is {}", batchSize);
+ }
+
+ Integer lingerMsConfigured = Tools.getIntegerProperty(properties, LINGER_MS);
+ if (lingerMsConfigured == null) {
+ lingerMs = DEFAULT_KAFKA_LINGER_MS;
+ log.info("Kafka lingerMs property is NOT configured, default value is {}", lingerMs);
+ } else {
+ lingerMs = lingerMsConfigured;
+ log.info("Configured. Kafka lingerMs is {}", lingerMs);
+ }
+
+ Integer memoryBufferConfigured = Tools.getIntegerProperty(properties, MEMORY_BUFFER);
+ if (memoryBufferConfigured == null) {
+ memoryBuffer = DEFAULT_KAFKA_MEMORY_BUFFER;
+ log.info("Kafka memory buffer property is NOT configured, default value is {}", memoryBuffer);
+ } else {
+ memoryBuffer = memoryBufferConfigured;
+ log.info("Configured. Kafka memory buffer is {}", memoryBuffer);
+ }
+
+ String keySerializerStr = Tools.get(properties, KEY_SERIALIZER);
+ keySerializer = keySerializerStr != null ? keySerializerStr : DEFAULT_KAFKA_KEY_SERIALIZER;
+ log.info("Configured, Kafka key serializer is {}", keySerializer);
+
+ String valueSerializerStr = Tools.get(properties, VALUE_SERIALIZER);
+ valueSerializer = valueSerializerStr != null ? valueSerializerStr : DEFAULT_KAFKA_VALUE_SERIALIZER;
+ log.info("Configured, Kafka value serializer is {}", valueSerializer);
}
}