diff options
Diffstat (limited to 'src/main/java/com/att/dmf/mr/metrics/publisher/impl')
4 files changed, 772 insertions, 0 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 new file mode 100644 index 0000000..08b2fd1 --- /dev/null +++ b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/CambriaBaseClient.java @@ -0,0 +1,100 @@ +/******************************************************************************* + * ============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.Logger; +//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 ( hosts, CambriaConstants.kStdCambriaServicePort, clientSignature, + CacheUse.NONE, 1, 1, TimeUnit.MILLISECONDS );*/ + + super(ConnectionType.HTTP, hosts, CambriaConstants.kStdCambriaServicePort, clientSignature, CacheUse.NONE, 1, 1L, TimeUnit.MILLISECONDS, 32, 32, 600000); + + //fLog = LoggerFactory.getLogger ( this.getClass().getName () ); + fLog = EELFManager.getInstance().getLogger(this.getClass().getName()); + //( 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<String> (); + for ( int i=0; i<a.length (); i++ ) + { + set.add ( a.getString ( i )); + } + return set; + } + /** + * @param log + */ + public void logTo ( EELFLogger log ) + { + fLog = log; + + //replaceLogger ( 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 new file mode 100644 index 0000000..7463700 --- /dev/null +++ b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/Clock.java @@ -0,0 +1,74 @@ +/******************************************************************************* + * ============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 new file mode 100644 index 0000000..ee56213 --- /dev/null +++ b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaConsumerImpl.java @@ -0,0 +1,169 @@ +/******************************************************************************* + * ============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 new file mode 100644 index 0000000..d8d8799 --- /dev/null +++ b/src/main/java/com/att/dmf/mr/metrics/publisher/impl/DMaaPCambriaSimplerBatchPublisher.java @@ -0,0 +1,429 @@ +/******************************************************************************* + * ============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.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) { + throw new RuntimeException(e); + } + } + + 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.size() > 0) { + 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.size() > 0) { + 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.size() < 1) { + 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 + + // final String serverCalculatedSignature = sha1HmacSigner.sign + // ("2015-09-21T11:38:19-0700", "iHAxArrj6Ve9JgmHvR077QiV"); + + 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); + // header("X-CambriaAuth", + // "2OH46YIWa329QpEF:"+serverCalculatedSignature). + // header("X-CambriaDate", "2015-09-21T11:38:19-0700"). + // post(Entity.json(baseStream.toByteArray())); + + getLog().info("Response received :: " + response.getStatus()); + getLog().info("Response received :: " + response.toString()); + + // code from REST Client Ends + + /* + * final JSONObject result = post ( url, contentType, + * baseStream.toByteArray(), true ); final String logLine = + * "cambria reply ok (" + (Clock.now()-startMs) + " ms):" + + * result.toString (); getLog().info ( logLine ); + */ + fPending.clear(); + return true; + } catch (IllegalArgumentException x) { + getLog().warn(x.getMessage(), x); + } + /* + * catch ( HttpObjectNotFoundException x ) { getLog().warn ( + * x.getMessage(), x ); } catch ( HttpException 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(); + } + } + +} |