diff options
Diffstat (limited to 'src/main/java/com/att/dmf/mr/metrics/publisher/impl')
4 files changed, 0 insertions, 765 deletions
diff --git a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/CambriaBaseClient.java b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/CambriaBaseClient.java deleted file mode 100644 index 84576fc..0000000 --- a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/CambriaBaseClient.java +++ /dev/null @@ -1,100 +0,0 @@ -/******************************************************************************* - * ============LICENSE_START======================================================= - * org.onap.dmaap - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * 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. - * ============LICENSE_END========================================================= - * - * ECOMP is a trademark and service mark of AT&T Intellectual Property. - * - *******************************************************************************/ -package com.att.dmf.mr.metrics.publisher.impl; - -import java.net.MalformedURLException; -import java.util.Collection; -import java.util.Set; -import java.util.TreeSet; -import java.util.concurrent.TimeUnit; - -import org.json.JSONArray; -import org.json.JSONException; - -import com.att.dmf.mr.constants.CambriaConstants; - -//import org.slf4j.LoggerFactory; -import com.att.eelf.configuration.EELFLogger; -import com.att.eelf.configuration.EELFManager; -import com.att.nsa.apiClient.http.CacheUse; -import com.att.nsa.apiClient.http.HttpClient; - -/** - * - * @author anowarul.islam - * - */ -public class CambriaBaseClient extends HttpClient implements com.att.dmf.mr.metrics.publisher.CambriaClient -{ - protected CambriaBaseClient ( Collection<String> hosts ) throws MalformedURLException - { - this ( hosts, null ); - } - - public CambriaBaseClient ( Collection<String> hosts, String clientSignature ) throws MalformedURLException - { - - - - super(ConnectionType.HTTP, hosts, CambriaConstants.kStdCambriaServicePort, clientSignature, CacheUse.NONE, 1, 1L, TimeUnit.MILLISECONDS, 32, 32, 600000); - - - fLog = EELFManager.getInstance().getLogger(this.getClass().getName()); - - } - - @Override - public void close () - { - } - - public Set<String> jsonArrayToSet ( JSONArray a ) throws JSONException - { - if ( a == null ) return null; - - final TreeSet<String> set = new TreeSet<>(); - for ( int i=0; i<a.length (); i++ ) - { - set.add ( a.getString ( i )); - } - return set; - } - /** - * @param log - */ - public void logTo ( EELFLogger log ) - { - fLog = log; - - - } - - public EELFLogger getLog () - { - return fLog; - } - - private EELFLogger fLog; - - - -} diff --git a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/Clock.java b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/Clock.java deleted file mode 100644 index 7463700..0000000 --- a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/Clock.java +++ /dev/null @@ -1,74 +0,0 @@ -/******************************************************************************* - * ============LICENSE_START======================================================= - * org.onap.dmaap - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * 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. - * ============LICENSE_END========================================================= - * - * ECOMP is a trademark and service mark of AT&T Intellectual Property. - * - *******************************************************************************/ -package com.att.dmf.mr.metrics.publisher.impl; - -/** - * - * This class maintains the system clocks - * @author nilanjana.maity - * - */ -public class Clock -{ - public synchronized static Clock getIt () - { - if ( sfClock == null ) - { - sfClock = new Clock (); - } - return sfClock; - } - - /** - * - * Get the system's current time in milliseconds. - * @return the current time - * - */ - public static long now () - { - return getIt().nowImpl (); - } - - /** - * Get current time in milliseconds - * @return current time in ms - */ - public long nowImpl () - { - return System.currentTimeMillis (); - } - - /** - * Initialize constructor - */ - public Clock () - { - } - - private static Clock sfClock = null; - - public synchronized static void register ( Clock testClock ) - { - sfClock = testClock; - } -} diff --git a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaConsumerImpl.java b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaConsumerImpl.java deleted file mode 100644 index ee56213..0000000 --- a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaConsumerImpl.java +++ /dev/null @@ -1,169 +0,0 @@ -/******************************************************************************* - * ============LICENSE_START======================================================= - * org.onap.dmaap - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * 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. - * ============LICENSE_END========================================================= - * - * ECOMP is a trademark and service mark of AT&T Intellectual Property. - * - *******************************************************************************/ -package com.att.dmf.mr.metrics.publisher.impl; - -import java.io.IOException; -import java.io.UnsupportedEncodingException; -import java.net.MalformedURLException; -import java.net.URLEncoder; -import java.util.Collection; -import java.util.LinkedList; -import java.util.List; - -import org.json.JSONArray; -import org.json.JSONException; -import org.json.JSONObject; - -import com.att.dmf.mr.metrics.publisher.CambriaPublisherUtility; -import com.att.nsa.apiClient.http.HttpException; -import com.att.nsa.apiClient.http.HttpObjectNotFoundException; - -import jline.internal.Log; - -/** - * - * @author anowarul.islam - * - */ -public class DMaaPCambriaConsumerImpl extends CambriaBaseClient - implements com.att.dmf.mr.metrics.publisher.CambriaConsumer { - private final String fTopic; - private final String fGroup; - private final String fId; - private final int fTimeoutMs; - private final int fLimit; - private final String fFilter; - - /** - * - * @param hostPart - * @param topic - * @param consumerGroup - * @param consumerId - * @param timeoutMs - * @param limit - * @param filter - * @param apiKey - * @param apiSecret - */ - public DMaaPCambriaConsumerImpl(Collection<String> hostPart, final String topic, final String consumerGroup, - final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret) throws MalformedURLException { - super(hostPart, topic + "::" + consumerGroup + "::" + consumerId); - - fTopic = topic; - fGroup = consumerGroup; - fId = consumerId; - fTimeoutMs = timeoutMs; - fLimit = limit; - fFilter = filter; - - setApiCredentials(apiKey, apiSecret); - } - - /** - * method converts String to list - * - * @param str - * @return - */ - public static List<String> stringToList(String str) { - final LinkedList<String> set = new LinkedList<String>(); - if (str != null) { - final String[] parts = str.trim().split(","); - for (String part : parts) { - final String trimmed = part.trim(); - if (trimmed.length() > 0) { - set.add(trimmed); - } - } - } - return set; - } - - @Override - public Iterable<String> fetch() throws IOException { - // fetch with the timeout and limit set in constructor - return fetch(fTimeoutMs, fLimit); - } - - @Override - public Iterable<String> fetch(int timeoutMs, int limit) throws IOException { - final LinkedList<String> msgs = new LinkedList<String>(); - - final String urlPath = createUrlPath(timeoutMs, limit); - - getLog().info("UEB GET " + urlPath); - try { - final JSONObject o = get(urlPath); - - if (o != null) { - final JSONArray a = o.getJSONArray("result"); - if (a != null) { - for (int i = 0; i < a.length(); i++) { - msgs.add(a.getString(i)); - } - } - } - } catch (HttpObjectNotFoundException e) { - // this can happen if the topic is not yet created. ignore. - Log.error("Failed due to topic is not yet created" + e); - } catch (JSONException e) { - // unexpected response - reportProblemWithResponse(); - Log.error("Failed due to jsonException", e); - } catch (HttpException e) { - throw new IOException(e); - } - - return msgs; - } - - public String createUrlPath(int timeoutMs, int limit) { - final StringBuilder url = new StringBuilder(CambriaPublisherUtility.makeConsumerUrl(fTopic, fGroup, fId)); - final StringBuilder adds = new StringBuilder(); - if (timeoutMs > -1) { - adds.append("timeout=").append(timeoutMs); - } - - if (limit > -1) { - if (adds.length() > 0) { - adds.append("&"); - } - adds.append("limit=").append(limit); - } - if (fFilter != null && fFilter.length() > 0) { - try { - if (adds.length() > 0) { - adds.append("&"); - } - adds.append("filter=").append(URLEncoder.encode(fFilter, "UTF-8")); - } catch (UnsupportedEncodingException e) { - Log.error("Failed due to UnsupportedEncodingException" + e); - } - } - if (adds.length() > 0) { - url.append("?").append(adds.toString()); - } - return url.toString(); - } - -} diff --git a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java deleted file mode 100644 index e9b1cdb..0000000 --- a/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java +++ /dev/null @@ -1,422 +0,0 @@ -/******************************************************************************* - * ============LICENSE_START======================================================= - * org.onap.dmaap - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * 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. - * ============LICENSE_END========================================================= - * - * ECOMP is a trademark and service mark of AT&T Intellectual Property. - * - *******************************************************************************/ -package com.att.dmf.mr.metrics.publisher.impl; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.OutputStream; -import java.net.MalformedURLException; -import java.nio.channels.NotYetConnectedException; -import java.util.Collection; -import java.util.LinkedList; -import java.util.List; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.zip.GZIPOutputStream; - -import javax.ws.rs.client.Client; -import javax.ws.rs.client.ClientBuilder; -import javax.ws.rs.client.Entity; -import javax.ws.rs.client.WebTarget; -import javax.ws.rs.core.Response; - -import com.att.ajsc.filemonitor.AJSCPropertiesMap; -import com.att.dmf.mr.constants.CambriaConstants; -import com.att.dmf.mr.metrics.publisher.CambriaPublisherUtility; - -/** - * - * class DMaaPCambriaSimplerBatchPublisher used to send the publish the messages - * in batch - * - * @author anowarul.islam - * - */ -public class DMaaPCambriaSimplerBatchPublisher extends CambriaBaseClient - implements com.att.dmf.mr.metrics.publisher.CambriaBatchingPublisher { - /** - * - * static inner class initializes with urls, topic,batchSize - * - * @author anowarul.islam - * - */ - public static class Builder { - public Builder() { - } - - /** - * constructor initialize with url - * - * @param baseUrls - * @return - * - */ - public Builder againstUrls(Collection<String> baseUrls) { - fUrls = baseUrls; - return this; - } - - /** - * constructor initializes with topics - * - * @param topic - * @return - * - */ - public Builder onTopic(String topic) { - fTopic = topic; - return this; - } - - /** - * constructor initilazes with batch size and batch time - * - * @param maxBatchSize - * @param maxBatchAgeMs - * @return - * - */ - public Builder batchTo(int maxBatchSize, long maxBatchAgeMs) { - fMaxBatchSize = maxBatchSize; - fMaxBatchAgeMs = maxBatchAgeMs; - return this; - } - - /** - * constructor initializes with compress - * - * @param compress - * @return - */ - public Builder compress(boolean compress) { - fCompress = compress; - return this; - } - - /** - * method returns DMaaPCambriaSimplerBatchPublisher object - * - * @return - */ - public DMaaPCambriaSimplerBatchPublisher build() { - - try { - return new DMaaPCambriaSimplerBatchPublisher(fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress); - } catch (MalformedURLException e) { - - NotYetConnectedException exception=new NotYetConnectedException(); - exception.setStackTrace(e.getStackTrace()); - - throw exception ; - - } - } - - private Collection<String> fUrls; - private String fTopic; - private int fMaxBatchSize = 100; - private long fMaxBatchAgeMs = 1000; - private boolean fCompress = false; - }; - - /** - * - * @param partition - * @param msg - */ - @Override - public int send(String partition, String msg) { - return send(new message(partition, msg)); - } - - /** - * @param msg - */ - @Override - public int send(message msg) { - final LinkedList<message> list = new LinkedList<message>(); - list.add(msg); - return send(list); - } - - /** - * @param msgs - */ - @Override - public synchronized int send(Collection<message> msgs) { - if (fClosed) { - throw new IllegalStateException("The publisher was closed."); - } - - for (message userMsg : msgs) { - fPending.add(new TimestampedMessage(userMsg)); - } - return getPendingMessageCount(); - } - - /** - * getPending message count - */ - @Override - public synchronized int getPendingMessageCount() { - return fPending.size(); - } - - /** - * - * @exception InterruptedException - * @exception IOException - */ - @Override - public void close() { - try { - final List<message> remains = close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); - if (remains.isEmpty()) { - getLog().warn("Closing publisher with " + remains.size() + " messages unsent. " - + "Consider using CambriaBatchingPublisher.close( long timeout, TimeUnit timeoutUnits ) to recapture unsent messages on close."); - } - } catch (InterruptedException e) { - getLog().warn("Possible message loss. " + e.getMessage(), e); - } catch (IOException e) { - getLog().warn("Possible message loss. " + e.getMessage(), e); - } - } - - /** - * @param time - * @param unit - */ - @Override - public List<message> close(long time, TimeUnit unit) throws IOException, InterruptedException { - synchronized (this) { - fClosed = true; - - // stop the background sender - fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); - fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); - fExec.shutdown(); - } - - final long now = Clock.now(); - final long waitInMs = TimeUnit.MILLISECONDS.convert(time, unit); - final long timeoutAtMs = now + waitInMs; - - while (Clock.now() < timeoutAtMs && getPendingMessageCount() > 0) { - send(true); - Thread.sleep(250); - } - // synchronizing the current object - synchronized (this) { - final LinkedList<message> result = new LinkedList<message>(); - fPending.drainTo(result); - return result; - } - } - - /** - * Possibly send a batch to the cambria server. This is called by the - * background thread and the close() method - * - * @param force - */ - private synchronized void send(boolean force) { - if (force || shouldSendNow()) { - if (!sendBatch()) { - getLog().warn("Send failed, " + fPending.size() + " message to send."); - - // note the time for back-off - fDontSendUntilMs = sfWaitAfterError + Clock.now(); - } - } - } - - /** - * - * @return - */ - private synchronized boolean shouldSendNow() { - boolean shouldSend = false; - if (fPending.isEmpty()) { - final long nowMs = Clock.now(); - - shouldSend = (fPending.size() >= fMaxBatchSize); - if (!shouldSend) { - final long sendAtMs = fPending.peek().timestamp + fMaxBatchAgeMs; - shouldSend = sendAtMs <= nowMs; - } - - // however, wait after an error - shouldSend = shouldSend && nowMs >= fDontSendUntilMs; - } - return shouldSend; - } - - /** - * - * @return - */ - private synchronized boolean sendBatch() { - // it's possible for this call to be made with an empty list. in this - // case, just return. - if (fPending.isEmpty()) { - return true; - } - - final long nowMs = Clock.now(); - final String url = CambriaPublisherUtility.makeUrl(fTopic); - - getLog().info("sending " + fPending.size() + " msgs to " + url + ". Oldest: " - + (nowMs - fPending.peek().timestamp) + " ms"); - - try { - - final ByteArrayOutputStream baseStream = new ByteArrayOutputStream(); - OutputStream os = baseStream; - if (fCompress) { - os = new GZIPOutputStream(baseStream); - } - for (TimestampedMessage m : fPending) { - os.write(("" + m.fPartition.length()).getBytes()); - os.write('.'); - os.write(("" + m.fMsg.length()).getBytes()); - os.write('.'); - os.write(m.fPartition.getBytes()); - os.write(m.fMsg.getBytes()); - os.write('\n'); - } - os.close(); - - final long startMs = Clock.now(); - - // code from REST Client Starts - - - - - Client client = ClientBuilder.newClient(); - String metricTopicname = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"metrics.send.cambria.topic"); - if (null==metricTopicname) { - - metricTopicname="msgrtr.apinode.metrics.dmaap"; - } - WebTarget target = client - .target("http://localhost:" + CambriaConstants.kStdCambriaServicePort); - target = target.path("/events/" + fTopic); - getLog().info("url : " + target.getUri().toString()); - // API Key - - Entity<byte[]> data = Entity.entity(baseStream.toByteArray(), "application/cambria"); - - Response response = target.request().post(data); - - getLog().info("Response received :: " + response.getStatus()); - getLog().info("Response received :: " + response.toString()); - - // code from REST Client Ends - - - fPending.clear(); - return true; - } catch (IllegalArgumentException x) { - getLog().warn(x.getMessage(), x); - } - - catch (IOException x) { - getLog().warn(x.getMessage(), x); - } - return false; - } - - private final String fTopic; - private final int fMaxBatchSize; - private final long fMaxBatchAgeMs; - private final boolean fCompress; - private boolean fClosed; - - private final LinkedBlockingQueue<TimestampedMessage> fPending; - private long fDontSendUntilMs; - private final ScheduledThreadPoolExecutor fExec; - - private static final long sfWaitAfterError = 1000; - - /** - * - * @param hosts - * @param topic - * @param maxBatchSize - * @param maxBatchAgeMs - * @param compress - */ - private DMaaPCambriaSimplerBatchPublisher(Collection<String> hosts, String topic, int maxBatchSize, - long maxBatchAgeMs, boolean compress) throws MalformedURLException { - - super(hosts); - - if (topic == null || topic.length() < 1) { - throw new IllegalArgumentException("A topic must be provided."); - } - - fClosed = false; - fTopic = topic; - fMaxBatchSize = maxBatchSize; - fMaxBatchAgeMs = maxBatchAgeMs; - fCompress = compress; - - fPending = new LinkedBlockingQueue<TimestampedMessage>(); - fDontSendUntilMs = 0; - - fExec = new ScheduledThreadPoolExecutor(1); - fExec.scheduleAtFixedRate(new Runnable() { - @Override - public void run() { - send(false); - } - }, 100, 50, TimeUnit.MILLISECONDS); - } - - /** - * - * - * @author anowarul.islam - * - */ - private static class TimestampedMessage extends message { - /** - * to store timestamp value - */ - public final long timestamp; - - /** - * constructor initialize with message - * - * @param m - * - */ - public TimestampedMessage(message m) { - super(m); - timestamp = Clock.now(); - } - } - -} |