log4j-flume-ng/src/main/java/org/apache/logging/log4j/flume/appender/FlumeAvroManager.java - logging-log4j2 - Git at Google

 /*
  * 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.logging.log4j.flume.appender;

 import java.util.Properties;
 import java.util.concurrent.TimeUnit;

 import org.apache.flume.Event;
 import org.apache.flume.api.RpcClient;
 import org.apache.flume.api.RpcClientFactory;
 import org.apache.logging.log4j.core.appender.AppenderLoggingException;
 import org.apache.logging.log4j.core.appender.ManagerFactory;

 /**
  * Manager for FlumeAvroAppenders.
  */
 public class FlumeAvroManager extends AbstractFlumeManager {

     private static final int MAX_RECONNECTS = 3;
     private static final int MINIMUM_TIMEOUT = 1000;

     private static AvroManagerFactory factory = new AvroManagerFactory();

     private final Agent[] agents;

     private final int batchSize;

     private final long delayNanos;
     private final int delayMillis;

     private final int retries;

     private final int connectTimeoutMillis;

     private final int requestTimeoutMillis;

     private final int current = 0;

     private volatile RpcClient rpcClient = null;

     private BatchEvent batchEvent = new BatchEvent();
     private long nextSend = 0;

     /**
      * Constructor
      * @param name The unique name of this manager.
      * @param agents An array of Agents.
      * @param batchSize The number of events to include in a batch.
      * @param retries The number of times to retry connecting before giving up.
      * @param connectTimeout The connection timeout in ms.
      * @param requestTimeout The request timeout in ms.
      *
      */
     protected FlumeAvroManager(final String name, final String shortName, final Agent[] agents, final int batchSize,
                                final int delayMillis, final int retries, final int connectTimeout, final int requestTimeout) {
         super(name);
         this.agents = agents;
         this.batchSize = batchSize;
         this.delayMillis = delayMillis;
         this.delayNanos = TimeUnit.MILLISECONDS.toNanos(delayMillis);
         this.retries = retries;
         this.connectTimeoutMillis = connectTimeout;
         this.requestTimeoutMillis = requestTimeout;
         this.rpcClient = connect(agents, retries, connectTimeout, requestTimeout);
     }

     /**
      * Returns a FlumeAvroManager.
      * @param name The name of the manager.
      * @param agents The agents to use.
      * @param batchSize The number of events to include in a batch.
      * @param delayMillis The number of milliseconds to wait before sending an incomplete batch.
      * @param retries The number of times to retry connecting before giving up.
      * @param connectTimeoutMillis The connection timeout in ms.
      * @param requestTimeoutMillis The request timeout in ms.
      * @return A FlumeAvroManager.
      */
     public static FlumeAvroManager getManager(final String name, final Agent[] agents, int batchSize, final int delayMillis,
                                               final int retries, final int connectTimeoutMillis, final int requestTimeoutMillis) {
         if (agents == null || agents.length == 0) {
             throw new IllegalArgumentException("At least one agent is required");
         }

         if (batchSize <= 0) {
             batchSize = 1;
         };
         final StringBuilder sb = new StringBuilder(name);
         sb.append(" FlumeAvro[");
         boolean first = true;
         for (final Agent agent : agents) {
             if (!first) {
                 sb.append(',');
             }
             sb.append(agent.getHost()).append(':').append(agent.getPort());
             first = false;
         }
         sb.append(']');
         return getManager(sb.toString(), factory,
                 new FactoryData(name, agents, batchSize, delayMillis, retries, connectTimeoutMillis, requestTimeoutMillis));
     }

     /**
      * Returns the agents.
      * @return The agent array.
      */
     public Agent[] getAgents() {
         return agents;
     }

     /**
      * Returns the index of the current agent.
      * @return The index for the current agent.
      */
     public int getCurrent() {
         return current;
     }

     public int getRetries() {
         return retries;
     }

     public int getConnectTimeoutMillis() {
         return connectTimeoutMillis;
     }

     public int getRequestTimeoutMillis() {
         return requestTimeoutMillis;
     }

     public int getBatchSize() {
         return batchSize;
     }

     public int getDelayMillis() {
         return delayMillis;
     }

     public void send(final BatchEvent events) {
         if (rpcClient == null) {
             synchronized (this) {
                 if (rpcClient == null) {
                     rpcClient = connect(agents, retries, connectTimeoutMillis, requestTimeoutMillis);
                 }
             }
         }

         if (rpcClient != null) {
             try {
                 LOGGER.trace("Sending batch of {} events", events.getEvents().size());
                 rpcClient.appendBatch(events.getEvents());
             } catch (final Exception ex) {
                 rpcClient.close();
                 rpcClient = null;
                 final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
                     agents[current].getPort();
                 LOGGER.warn(msg, ex);
                 throw new AppenderLoggingException("No Flume agents are available");
             }
         }  else {
             final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
                 agents[current].getPort();
             LOGGER.warn(msg);
             throw new AppenderLoggingException("No Flume agents are available");
         }
     }

     @Override
     public void send(final Event event)  {
         if (batchSize == 1) {
             if (rpcClient == null) {
                 rpcClient = connect(agents, retries, connectTimeoutMillis, requestTimeoutMillis);
             }

             if (rpcClient != null) {
                 try {
                     rpcClient.append(event);
                 } catch (final Exception ex) {
                     rpcClient.close();
                     rpcClient = null;
                     final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
                             agents[current].getPort();
                     LOGGER.warn(msg, ex);
                     throw new AppenderLoggingException("No Flume agents are available");
                 }
             } else {
                 final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
                         agents[current].getPort();
                 LOGGER.warn(msg);
                 throw new AppenderLoggingException("No Flume agents are available");
             }
         } else {
             int eventCount;
             BatchEvent batch = null;
             synchronized(batchEvent) {
                 batchEvent.addEvent(event);
                 eventCount = batchEvent.size();
                 long now = System.nanoTime();
                 if (eventCount == 1) {
                     nextSend = now + delayNanos;
                 }
                 if (eventCount >= batchSize || now >= nextSend) {
                     batch = batchEvent;
                     batchEvent = new BatchEvent();
                 }
             }
             if (batch != null) {
                 send(batch);
             }
         }
     }

     /**
      * There is a very good chance that this will always return the first agent even if it isn't available.
      * @param agents The list of agents to choose from
      * @return The FlumeEventAvroServer.
      */
     private RpcClient connect(final Agent[] agents, int retries, final int connectTimeoutMillis, final int requestTimeoutMillis) {
         try {
             final Properties props = new Properties();

             props.put("client.type", "default_failover");

             int agentCount = 1;
             final StringBuilder sb = new StringBuilder();
             for (final Agent agent : agents) {
                 if (sb.length() > 0) {
                     sb.append(' ');
                 }
                 final String hostName = "host" + agentCount++;
                 props.put("hosts." + hostName, agent.getHost() + ':' + agent.getPort());
                 sb.append(hostName);
             }
             props.put("hosts", sb.toString());
             if (batchSize > 0) {
                 props.put("batch-size", Integer.toString(batchSize));
             }
             if (retries > 1) {
                 if (retries > MAX_RECONNECTS) {
                     retries = MAX_RECONNECTS;
                 }
                 props.put("max-attempts", Integer.toString(retries * agents.length));
             }
             if (requestTimeoutMillis >= MINIMUM_TIMEOUT) {
                 props.put("request-timeout", Integer.toString(requestTimeoutMillis));
             }
             if (connectTimeoutMillis >= MINIMUM_TIMEOUT) {
                 props.put("connect-timeout", Integer.toString(connectTimeoutMillis));
             }
             return RpcClientFactory.getInstance(props);
         } catch (final Exception ex) {
             LOGGER.error("Unable to create Flume RPCClient: {}", ex.getMessage());
             return null;
         }
     }

     @Override
     protected boolean releaseSub(final long timeout, final TimeUnit timeUnit) {
     	boolean closed = true;
         if (rpcClient != null) {
             try {
                 synchronized(this) {
                     try {
                         if (batchSize > 1 && batchEvent.getEvents().size() > 0) {
                             send(batchEvent);
                         }
                     } catch (final Exception ex) {
                         LOGGER.error("Error sending final batch: {}", ex.getMessage());
                         closed = false;
                     }
                 }
                 rpcClient.close();
             } catch (final Exception ex) {
                 LOGGER.error("Attempt to close RPC client failed", ex);
                 closed = false;
             }
         }
         rpcClient = null;
         return closed;
     }

     /**
      * Factory data.
      */
     private static class FactoryData {
         private final String name;
         private final Agent[] agents;
         private final int batchSize;
         private final int delayMillis;
         private final int retries;
         private final int conntectTimeoutMillis;
         private final int requestTimeoutMillis;

         /**
          * Constructor.
          * @param name The name of the Appender.
          * @param agents The agents.
          * @param batchSize The number of events to include in a batch.
          */
         public FactoryData(final String name, final Agent[] agents, final int batchSize, final int delayMillis,
                 final int retries, final int connectTimeoutMillis, final int requestTimeoutMillis) {
             this.name = name;
             this.agents = agents;
             this.batchSize = batchSize;
             this.delayMillis = delayMillis;
             this.retries = retries;
             this.conntectTimeoutMillis = connectTimeoutMillis;
             this.requestTimeoutMillis = requestTimeoutMillis;
         }
     }

     /**
      * Avro Manager Factory.
      */
     private static class AvroManagerFactory implements ManagerFactory<FlumeAvroManager, FactoryData> {

         /**
          * Create the FlumeAvroManager.
          * @param name The name of the entity to manage.
          * @param data The data required to create the entity.
          * @return The FlumeAvroManager.
          */
         @Override
         public FlumeAvroManager createManager(final String name, final FactoryData data) {
             try {

                 return new FlumeAvroManager(name, data.name, data.agents, data.batchSize, data.delayMillis,
                         data.retries, data.conntectTimeoutMillis, data.requestTimeoutMillis);
             } catch (final Exception ex) {
                 LOGGER.error("Could not create FlumeAvroManager", ex);
             }
             return null;
         }
     }

 }
	/*
	* 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.logging.log4j.flume.appender;

	import java.util.Properties;
	import java.util.concurrent.TimeUnit;

	import org.apache.flume.Event;
	import org.apache.flume.api.RpcClient;
	import org.apache.flume.api.RpcClientFactory;
	import org.apache.logging.log4j.core.appender.AppenderLoggingException;
	import org.apache.logging.log4j.core.appender.ManagerFactory;

	/**
	* Manager for FlumeAvroAppenders.
	*/
	public class FlumeAvroManager extends AbstractFlumeManager {

	private static final int MAX_RECONNECTS = 3;
	private static final int MINIMUM_TIMEOUT = 1000;

	private static AvroManagerFactory factory = new AvroManagerFactory();

	private final Agent[] agents;

	private final int batchSize;

	private final long delayNanos;
	private final int delayMillis;

	private final int retries;

	private final int connectTimeoutMillis;

	private final int requestTimeoutMillis;

	private final int current = 0;

	private volatile RpcClient rpcClient = null;

	private BatchEvent batchEvent = new BatchEvent();
	private long nextSend = 0;

	/**
	* Constructor
	* @param name The unique name of this manager.
	* @param agents An array of Agents.
	* @param batchSize The number of events to include in a batch.
	* @param retries The number of times to retry connecting before giving up.
	* @param connectTimeout The connection timeout in ms.
	* @param requestTimeout The request timeout in ms.
	*
	*/
	protected FlumeAvroManager(final String name, final String shortName, final Agent[] agents, final int batchSize,
	final int delayMillis, final int retries, final int connectTimeout, final int requestTimeout) {
	super(name);
	this.agents = agents;
	this.batchSize = batchSize;
	this.delayMillis = delayMillis;
	this.delayNanos = TimeUnit.MILLISECONDS.toNanos(delayMillis);
	this.retries = retries;
	this.connectTimeoutMillis = connectTimeout;
	this.requestTimeoutMillis = requestTimeout;
	this.rpcClient = connect(agents, retries, connectTimeout, requestTimeout);
	}

	/**
	* Returns a FlumeAvroManager.
	* @param name The name of the manager.
	* @param agents The agents to use.
	* @param batchSize The number of events to include in a batch.
	* @param delayMillis The number of milliseconds to wait before sending an incomplete batch.
	* @param retries The number of times to retry connecting before giving up.
	* @param connectTimeoutMillis The connection timeout in ms.
	* @param requestTimeoutMillis The request timeout in ms.
	* @return A FlumeAvroManager.
	*/
	public static FlumeAvroManager getManager(final String name, final Agent[] agents, int batchSize, final int delayMillis,
	final int retries, final int connectTimeoutMillis, final int requestTimeoutMillis) {
	if (agents == null \|\| agents.length == 0) {
	throw new IllegalArgumentException("At least one agent is required");
	}

	if (batchSize <= 0) {
	batchSize = 1;
	};
	final StringBuilder sb = new StringBuilder(name);
	sb.append(" FlumeAvro[");
	boolean first = true;
	for (final Agent agent : agents) {
	if (!first) {
	sb.append(',');
	}
	sb.append(agent.getHost()).append(':').append(agent.getPort());
	first = false;
	}
	sb.append(']');
	return getManager(sb.toString(), factory,
	new FactoryData(name, agents, batchSize, delayMillis, retries, connectTimeoutMillis, requestTimeoutMillis));
	}

	/**
	* Returns the agents.
	* @return The agent array.
	*/
	public Agent[] getAgents() {
	return agents;
	}

	/**
	* Returns the index of the current agent.
	* @return The index for the current agent.
	*/
	public int getCurrent() {
	return current;
	}

	public int getRetries() {
	return retries;
	}

	public int getConnectTimeoutMillis() {
	return connectTimeoutMillis;
	}

	public int getRequestTimeoutMillis() {
	return requestTimeoutMillis;
	}

	public int getBatchSize() {
	return batchSize;
	}

	public int getDelayMillis() {
	return delayMillis;
	}

	public void send(final BatchEvent events) {
	if (rpcClient == null) {
	synchronized (this) {
	if (rpcClient == null) {
	rpcClient = connect(agents, retries, connectTimeoutMillis, requestTimeoutMillis);
	}
	}
	}

	if (rpcClient != null) {
	try {
	LOGGER.trace("Sending batch of {} events", events.getEvents().size());
	rpcClient.appendBatch(events.getEvents());
	} catch (final Exception ex) {
	rpcClient.close();
	rpcClient = null;
	final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
	agents[current].getPort();
	LOGGER.warn(msg, ex);
	throw new AppenderLoggingException("No Flume agents are available");
	}
	} else {
	final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
	agents[current].getPort();
	LOGGER.warn(msg);
	throw new AppenderLoggingException("No Flume agents are available");
	}
	}

	@Override
	public void send(final Event event) {
	if (batchSize == 1) {
	if (rpcClient == null) {
	rpcClient = connect(agents, retries, connectTimeoutMillis, requestTimeoutMillis);
	}

	if (rpcClient != null) {
	try {
	rpcClient.append(event);
	} catch (final Exception ex) {
	rpcClient.close();
	rpcClient = null;
	final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
	agents[current].getPort();
	LOGGER.warn(msg, ex);
	throw new AppenderLoggingException("No Flume agents are available");
	}
	} else {
	final String msg = "Unable to write to " + getName() + " at " + agents[current].getHost() + ':' +
	agents[current].getPort();
	LOGGER.warn(msg);
	throw new AppenderLoggingException("No Flume agents are available");
	}
	} else {
	int eventCount;
	BatchEvent batch = null;
	synchronized(batchEvent) {
	batchEvent.addEvent(event);
	eventCount = batchEvent.size();
	long now = System.nanoTime();
	if (eventCount == 1) {
	nextSend = now + delayNanos;
	}
	if (eventCount >= batchSize \|\| now >= nextSend) {
	batch = batchEvent;
	batchEvent = new BatchEvent();
	}
	}
	if (batch != null) {
	send(batch);
	}
	}
	}

	/**
	* There is a very good chance that this will always return the first agent even if it isn't available.
	* @param agents The list of agents to choose from
	* @return The FlumeEventAvroServer.
	*/
	private RpcClient connect(final Agent[] agents, int retries, final int connectTimeoutMillis, final int requestTimeoutMillis) {
	try {
	final Properties props = new Properties();

	props.put("client.type", "default_failover");

	int agentCount = 1;
	final StringBuilder sb = new StringBuilder();
	for (final Agent agent : agents) {
	if (sb.length() > 0) {
	sb.append(' ');
	}
	final String hostName = "host" + agentCount++;
	props.put("hosts." + hostName, agent.getHost() + ':' + agent.getPort());
	sb.append(hostName);
	}
	props.put("hosts", sb.toString());
	if (batchSize > 0) {
	props.put("batch-size", Integer.toString(batchSize));
	}
	if (retries > 1) {
	if (retries > MAX_RECONNECTS) {
	retries = MAX_RECONNECTS;
	}
	props.put("max-attempts", Integer.toString(retries * agents.length));
	}
	if (requestTimeoutMillis >= MINIMUM_TIMEOUT) {
	props.put("request-timeout", Integer.toString(requestTimeoutMillis));
	}
	if (connectTimeoutMillis >= MINIMUM_TIMEOUT) {
	props.put("connect-timeout", Integer.toString(connectTimeoutMillis));
	}
	return RpcClientFactory.getInstance(props);
	} catch (final Exception ex) {
	LOGGER.error("Unable to create Flume RPCClient: {}", ex.getMessage());
	return null;
	}
	}

	@Override
	protected boolean releaseSub(final long timeout, final TimeUnit timeUnit) {
	boolean closed = true;
	if (rpcClient != null) {
	try {
	synchronized(this) {
	try {
	if (batchSize > 1 && batchEvent.getEvents().size() > 0) {
	send(batchEvent);
	}
	} catch (final Exception ex) {
	LOGGER.error("Error sending final batch: {}", ex.getMessage());
	closed = false;
	}
	}
	rpcClient.close();
	} catch (final Exception ex) {
	LOGGER.error("Attempt to close RPC client failed", ex);
	closed = false;
	}
	}
	rpcClient = null;
	return closed;
	}

	/**
	* Factory data.
	*/
	private static class FactoryData {
	private final String name;
	private final Agent[] agents;
	private final int batchSize;
	private final int delayMillis;
	private final int retries;
	private final int conntectTimeoutMillis;
	private final int requestTimeoutMillis;

	/**
	* Constructor.
	* @param name The name of the Appender.
	* @param agents The agents.
	* @param batchSize The number of events to include in a batch.
	*/
	public FactoryData(final String name, final Agent[] agents, final int batchSize, final int delayMillis,
	final int retries, final int connectTimeoutMillis, final int requestTimeoutMillis) {
	this.name = name;
	this.agents = agents;
	this.batchSize = batchSize;
	this.delayMillis = delayMillis;
	this.retries = retries;
	this.conntectTimeoutMillis = connectTimeoutMillis;
	this.requestTimeoutMillis = requestTimeoutMillis;
	}
	}

	/**
	* Avro Manager Factory.
	*/
	private static class AvroManagerFactory implements ManagerFactory<FlumeAvroManager, FactoryData> {

	/**
	* Create the FlumeAvroManager.
	* @param name The name of the entity to manage.
	* @param data The data required to create the entity.
	* @return The FlumeAvroManager.
	*/
	@Override
	public FlumeAvroManager createManager(final String name, final FactoryData data) {
	try {

	return new FlumeAvroManager(name, data.name, data.agents, data.batchSize, data.delayMillis,
	data.retries, data.conntectTimeoutMillis, data.requestTimeoutMillis);
	} catch (final Exception ex) {
	LOGGER.error("Could not create FlumeAvroManager", ex);
	}
	return null;
	}
	}

	}