retel-io/ari-proxy

Ari-proxy connects Asterisk, an open source communication server, to the Apache Kafka distributed streaming platform.

ari-proxy

Ari-proxy connects Asterisk, an open source communication server, to the Apache Kafka distributed streaming platform.

Table of contents

1. Abstract
2. Getting started
3. Metrics
4. Compatibility
5. Contributing & feedback
6. Credit & License
7. Acknowledgements

Abstract

The motivation to create ari-proxy arose from the need to build distributed and resilient telephony services scaling up to millions of active users. Ari-proxy makes use of Kafka’s built-in routing concepts to ensure consistency of message streams and proper dispatching to the call-controller, the application implementing the service logic.

Please see docs/concepts.md for details on the concepts of message routing and session mapping.

Getting started

In order to operate ari-proxy, make sure you have a running instance of both Asterisk and Kafka server.

Prerequisites

ari-proxy is written in Java, so you should install and setup Java before continuing. The project is managed by maven, which requires you to install maven as well.

Building

Build the fat jar in target/:

mvn package

Configuration

ari-proxy expects the following configuration files, which should be passed to the jvm when running the fat-jar:

config	optional	purpose
service.conf	no	configure the service, see our template: service.conf.sample
log4j2.xml	yes	configure logging (if not specified, a bundled config will be used logging to STDOUT only)
jolokia.properties	no	configure jolokia agent properties, see our template: jolokia.properties.sample

Running

Run the fat jar:

java -Dconfig.file=/path/to/service.conf [-Dlog4j.configurationFile=/path/to/log4j2.xml] -jar target/ari-proxy-1.3.0-fat.jar

Persistence-store

There are two ways to persist the in-memory data storage (Asterisk Object ID -> Kafka Routing Key).

First there is Redis (default). The redis needs to be configured in service.conf file in order to be able to connect. Also a keyspace needs to be configured. Second possibility to store the data is Cassandra. This is a more robust but more complex configuration as it provides HA and better scalability. You need to configure the Cassandra nodes in the "datastax" section of "service.conf"

In order to choose one option you need to enable the one or the other backend by set the parameter persistence-store to one of these values:

• "io.retel.ariproxy.persistence.plugin.CassandraPersistenceStore"

• "io.retel.ariproxy.persistence.plugin.RedisPersistenceStore"

In case you want to use Cassandra you need to create a keyspace and table in Cassandra. This snippet might help to create one. (please adapt replication factor and names according your setup)

CREATE KEYSPACE retel with replication = {'class':'SimpleStrategy','replication_factor':1};

USE retel;

CREATE TABLE retel (
"key" text primary key,
"value" text
);

Hint: do not forget some kind of housekeeping by adding TTL or a cleanup job.

Monitoring

Health-Check

There are three routes to check for service health:

• /health
• /health/smoke
• /health/backing-services

/health is meant to check if the service itself is healthy in terms of JVM,Akka etc. settings. Yet to be implemented correctly.

/health/smoke is used for deployment to check if the service is started. Only checks for open port.

/health/backing-services checks all services that are necessary for running the application e.g. Kafka, Asterisk, Redis/Cassandra.

Metrics

Ari-proxy provides service specific metrics using the micrometer framework which are available via JMX or HTTP.

For further details see: Metrics

Compatibility

We aim for compatibility with the latest stable release of

• Java OpenJDK 17
• Asterisk
• the utilized Akka Modules

Contributing & feedback

To report a bug or make a request for new features, use the Issues Page in the ari-proxy Github project. We welcome any contributions. Please see the Developer Guidelines for instructions on how to submit changes.

Credit & License

ari-proxy is maintained by the folks at sipgate and licensed under the terms of the AGPL license.

Maintainers of this repository:

• Jöran @vinzens
• Maya @ironmaya
• Mia @mia-krause
• Sven @SvenKube
• Max @IllTemperedMax

Please refer to the Git commit log for a complete list of contributors.

Acknowledgements

ari-proxy is not the first of its kind. This project was inspired by the concepts underlying both go-ari-proxy by N-Visible as well as ari-proxy by CyCore Systems.

ari-proxy's Issues

java.lang.RuntimeException: Failed to extract resourceId from command

Bug Report

corresponding commit-id: latest of master branch
asterisk version: 17.9.4
kafka version: 2.7
java version: openjdk8

setup description

Docker Compose configuration with following nodes:

• Zookeeper
• 3 Kafka nodes
• Redis node
• Asterisk node
• ari-proxy node

failure description

For testing, I wanted to play an audio file on an incoming call. Therefore I answer the incoming call on STASIS_START, using a randomly generated command_id, with the following:

from kafka import KafkaConsumer, KafkaProducer


producer = KafkaProducer(bootstrap_servers=['localhost:9093'], value_serializer=lambda v: json.dumps(v).encode('utf-8'))

producer.send('ari-commands-topic', {
	"callContext":call_context,
	"commandId": command_id,
	"ariCommand": {
		"method": "POST",
		"url": f"/channels/{ channel_id }/answer"
	}
})

Which works fine. Next I react to the RESPONSE of the command above with the following, also using another randomly generated command_id:

producer.send('ari-commands-topic', {
	"callContext": call_context,
	"commandId": command_id,
	"ariCommand": {
		"method": "POST",
		"url": f"/channels/{ channel_id }/play",
		"body": {
			"channelId": channel_id, # not sure if required
			"media": "sound:https://example.com/audio.sln16"
		}
	}
})

This command lets ari-proxy fail with the following stacktrace:

2021-08-09/16:56:50,120/CEST [INFO] [ari-app-akka.actor.default-dispatcher-6] Materializer - [>>>   ARI RESPONSE] Element: {"type":"RESPONSE","commandsTopic":"ari-commands-topic","payload":{"status_code":204},"callContext":"c72c970c-9c5b-4722-817b-43e15a22045d","resources":[{"type":"CHANNEL","id":"1628521009.17"}],"commandId":"test-1234-26","commandRequest":{"method":"POST","url":"/channels/1628521009.17/answer"}}
2021-08-09/16:56:52,181/CEST [INFO] [ari-app-akka.actor.default-dispatcher-7] Materializer - [>>>   ARI COMMAND] Element: {"callContext": "c72c970c-9c5b-4722-817b-43e15a22045d", "commandId": "test-1234-46", "ariCommand": {"method": "POST", "url": "/channels/1628521009.17/play", "body": {"channelId": "1628521009.17", "media": "https://example.com/audio.sln16"}}}
2021-08-09/16:56:52,185/CEST [ERROR] [ari-app-akka.actor.default-dispatcher-7] ActorSystemImpl - Error in some stage; restarting stream ...
java.lang.IllegalStateException: java.lang.RuntimeException: Failed to extract resourceId from command 'AriCommand[body={"channelId":"1628521009.17","media":"https://example.com/audio.sln16"},method=POST,url=/channels/1628521009.17/play]'
	at io.retel.ariproxy.boundary.commandsandresponses.AriCommandResponseKafkaProcessor.lambda$null$7(AriCommandResponseKafkaProcessor.java:125)
	at io.vavr.control.Try.onFailure(Try.java:659)
	at io.retel.ariproxy.boundary.commandsandresponses.AriCommandResponseKafkaProcessor.lambda$run$8b4133d7$1(AriCommandResponseKafkaProcessor.java:123)
	at akka.stream.javadsl.Source.$anonfun$map$1(Source.scala:1969)
	at akka.stream.impl.fusing.Map$$anon$1.onPush(Ops.scala:52)
	at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:542)
	at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:423)
	at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:650)
	at akka.stream.impl.fusing.GraphInterpreterShell$AsyncInput.execute(ActorGraphInterpreter.scala:521)
	at akka.stream.impl.fusing.GraphInterpreterShell.processEvent(ActorGraphInterpreter.scala:625)
	at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:800)
	at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$shortCircuitBatch(ActorGraphInterpreter.scala:787)
	at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:819)
	at akka.actor.Actor.aroundReceive(Actor.scala:537)
	at akka.actor.Actor.aroundReceive$(Actor.scala:535)
	at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:716)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
	at akka.actor.ActorCell.invoke(ActorCell.scala:548)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
	at akka.dispatch.Mailbox.run(Mailbox.scala:231)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
Caused by: java.lang.RuntimeException: Failed to extract resourceId from command 'AriCommand[body={"channelId":"1628521009.17","media":"https://example.com/audio.sln16"},method=POST,url=/channels/1628521009.17/play]'
	at io.retel.ariproxy.boundary.commandsandresponses.AriCommandResponseProcessing.registerCallContext(AriCommandResponseProcessing.java:32)
	at io.retel.ariproxy.boundary.commandsandresponses.AriCommandResponseKafkaProcessor.lambda$run$8b4133d7$1(AriCommandResponseKafkaProcessor.java:119)
	... 22 more

I did some digging in the code, but could not find anything helpful about resourceId.
Am I missing something or how can I investigate this further?

expected behaviour description

Should play the provided audio file.

Thanks a lot for your great code base and your help!

Client example

Hello. Thank you for providing ari-proxy to the community. I need help to get started. Could you direct me to a java sample client program, preferably using akka, that connects to kafka and retrieves the Stasis start event and sends a few simple commands to Asterisk? For example Answer, then play a media file, and then Hangup. Many thanks.

config valitation

So far there is no validation of the config.
The Ari-proxy does read the config and starts even though the config is invalid and the ari-proxy is not able to connect since there is an invalid connection for asterisk and kafka and/or the connection is not working.
If the ari-proxy is not able to function properly it must not start and stay started. A problem in startup must be logged at least to stdout (if logger config is broken) and the service has to terminate.

asterisk 16 new Stasis Message

Bug Report

corresponding commit-id: latest

asterisk version: 16

kafka version:
not involved

setup description

Basic Call from Application

failure description

there is a non implemented Stasis Message which is comming from asterisk and which is not implmented. In this case "STASISSTATUS"

expected behaviour description

Should be able to extract recourceID out of all messages

2018-10-09/19:05:59,241/CEST [ERROR] [ari-proxy-for-some-stasis-app-akka.actor.default-dispatcher-41] ActorSystemImpl - java.lang.RuntimeException: Failed to extract resourceId from body={"variable":"STASISSTATUS","value":"SUCCESS","type":"ChannelVarset","timestamp":"2018-10-09T19:05:59.214+0200","channel":{"id":"83374320-dac0-4010-9ad6-c51e30fd93c0","name":"PJSIP/101-00000054","state":"Up","caller":{"name":"Temple Morgan","number":"101"},"connected":{"name":"","number":""},"accountcode":"","dialplan":{"context":"foo","exten":"s","priority":1},"creationtime":"2018-10-09T19:05:52.489+0200","language":"en","channelvars":{"X-AUTHDOMAIN":""}},"asterisk_id":"e4:b3:18:51:7d:4a","application":"application-name"}

Syntax of Call Controller (ari-commands-topic)

Facing issue in generating calls through call controller.

We tried two commands through ari-proxy:

1. GET channel
   

2. DELETE channel

Retry of http command if socket breaks

In the AriCommandResponseKafkaProcessor.class there is a step in order to place a raw http request to the asterisk Web socket. We have seen broken Web sockets and the commands could not be sent to the asterisk. In order to provide more stability we have to react on a broken web socket in this case and do a reconnect / retry of the command.
We haven't figured a way to reproduce this issue on purpose but it occur occasionally.

default-http-port in service.config

Feature Request

there is a default-http-port in application.conf which is bundled. There is no way to have the port configured by external config (service.conf)

proposal

use a field in service.conf in order to configure the port

current behaviour

no change possible without changing the application.conf

desired behaviour

port parameter in service.conf

use case / why is this important?

to have a more customization of configuration if the ports needs to be changed.

zookeeper support

Feature Request

I would like to suggest service discovery via zookeeper

proposal

current behaviour

currently the only support for a kafka cluster seems to be via a comma separated list which requires hard-coding the hostname:ports

desired behaviour

The desired behavior would for ari-proxy to pull from zookeeper the list of brokers

use case / why is this important?

one specific case would be when running a kafka cluster on docker, ports are assigned dynamically which makes it difficult to configure ari-proxy properly

How to import ari-proxy into Eclipse?

We wrote an Eclipse plugin and are now collecting a database of large Akka Java Maven projects. I tried to import ari-proxy into Eclipse, but I couldn't compile it and run the tests. I Just imported the ari-proxy as an existing maven project in eclipse but it shows 50 errors like the below samples:

I tried changing the java version or maven version, adding that plugin to maven dependencies, and following all of the StackOverflow suggestions, such as removing the.m2 folder, but none of them worked.

Is ari-proxy Eclipse compatible? Is it possible to run their test classes inside Eclipse? If so, what procedure should I follow?

Sample config is invalid

Bug Report

corresponding commit-id:latest
asterisk version:latest
kafka version:2.0
java version:8

setup description

failure description

When running the application via

java -Dconfig.file=service.conf  -jar target/ari-proxy-0.9.0-fat.jar 

, it exits saying no config websocket-uri is found , on closer inspection the ServiceConfig class doesnt have a asterisk-uri field only a websocket-uri field.
It would seem to be that the sample config file is inaccurate

expected behaviour description

to run