How to reduce mqtt data from shelly 3em with the help of "Final Aggregrator Plugin"

Telegraf
21

… and my telegraf.conf file looks like this

[global_tags]

[agent]

interval = “10s”

round_interval = true

metric_batch_size = 1000

metric_buffer_limit = 10000

collection_jitter = “0s”

debug = true

flush_interval = “10s”

flush_jitter = “0s”

precision = “”

#===================================================================================

[[outputs.file]]
files = [“stdout”]

#output configuration
#database “PV_data”

#measurement “grid” >>> Bezug / Einspeisung vom Shelly 3 EM

[[outputs.influxdb]]
urls = [“http://raspberrypi:8086”]
database = “PV_data”
name_override = “grid”
[outputs.influxdb.tagpass]
topic=[“shellies/em3/emeter/0/power”,“shellies/em3/emeter/1/power”,“shellies/em3/emeter/2/power”]

#measurement “w_garten” >>> Wintergarten

[[outputs.influxdb]]
urls = [“http://raspberrypi:8086”]
database = “PV_data”
name_override = “w_garten”
[outputs.influxdb.tagpass]
topic=[“tele/solar/SENSOR”]

#measurement “stall” für >>> Stall
[[outputs.influxdb]]
urls = [“http://raspberrypi:8086”]
database = “PV_data”
name_override = “stall”
[outputs.influxdb.tagpass]
topic=[“tele/Sonoff_PV001/SENSOR”]

#measurement “stall2” für >>> Stall2
[[outputs.influxdb]]
urls = [“http://raspberrypi:8086”]
database = “PV_data”
name_override = “stall2”
[outputs.influxdb.tagpass]
topic=[“tele/solar2/SENSOR”]

[[inputs.mqtt_consumer]]
servers = [“tcp://raspberrypi:1883”]
topics = [“tele/solar/SENSOR”,“tele/Sonoff_PV001/SENSOR”,“tele/solar2/SENSOR”]
#unnötige Messwerte werden nicht in der Datenbank aufgezeichnet -----------3.2.21 TG
fielddrop = [“ENERGY_ApparentPower”,“_Factor","_Period”,“*_ReactivePower”]
data_format = “json”

#topic von evcc System, zeigt ob Einspeisung oder Netzbezug erfolgt…25.10.22 TG
[[inputs.mqtt_consumer]]
servers = [“tcp://raspberrypi:1883”]
topics = [“shellies/em3/emeter/0/power”,“shellies/em3/emeter/1/power”,“shellies/em3/emeter/2/power”]
data_format = “value”
data_type = “float”

#shellies data weniger oft in influxdb schreiben

[[aggregators.final]]
namepass = [“grid”]
period = “30s”
drop_original = true
output_strategy = “periodic”

22

Note the name of the metric is mqtt_consumer. So for namepass, you use “mqtt_consumer”, not shellies or the database name.

23

Ok, but the problem with this is, that also the other inputs (3 pieces of Sonoff Pow R2) share the same name as mqtt_consumer …

[[inputs.mqtt_consumer]]
servers = [“tcp://raspberrypi:1883”]
topics = [“tele/solar/SENSOR”,“tele/Sonoff_PV001/SENSOR”,“tele/solar2/SENSOR”]
#unnötige Messwerte werden nicht in der Datenbank aufgezeichnet -----------3.2.21 TG
fielddrop = [“ENERGY_ApparentPower”,“_Factor"," _Period”,“*_ReactivePower”]
data_format = “json”

24

And what i see, that still every metrics is written to influxdb (every 2 or 3 sec), but in 30 sec interval then a “value_final”
… most probably from the “Final Aggregrator Plugin” function

25

Then use a name_override for the one mqtt_consumer to rename it and update the name pass in the aggregator: telegraf/docs/CONFIGURATION.md at master · influxdata/telegraf · GitHub

The other option is to use tagpass instead of namepass and only let things with the specific topics you want.

26

Ok, wonderful, with name_override only the choosen ones are selected for processing in “Final Aggregator Plugin”

Is there a solution to suppress the unwanted metrics … so that only every 30s metrics are written to the influxdb ??

27

You lost me :slight_smile: isn’t that the whole point of using the final aggregator? It only sends metrics every 30 seconds…

28

oh yes, but my setup still has some metrics inside the 30 sec

select * from grid order by time desc Limit 20
name: grid
time host topic value value_final

2024-03-08T16:58:26.515196334Z /em3/emeter/2/power 283.01
2024-03-08T16:58:26.515057064Z /em3/emeter/1/power 231.02
2024-03-08T16:58:26.514988783Z /em3/emeter/0/power 345.33
2024-03-08T16:58:26.514526234Z /em3/emeter/2/power 283.01
2024-03-08T16:58:26.51417384Z /em3/emeter/1/power 231.02
2024-03-08T16:58:26.513779519Z /em3/emeter/0/power 345.33
2024-03-08T16:58:19.517166389Z /em3/emeter/2/power 241.09
2024-03-08T16:58:19.516997953Z /em3/emeter/1/power 229.58
2024-03-08T16:58:19.516916495Z /em3/emeter/0/power 346.71
2024-03-08T16:58:08.509439936Z /em3/emeter/2/power 238.61
2024-03-08T16:58:08.509079782Z /em3/emeter/1/power 241.24
2024-03-08T16:58:08.508999783Z /em3/emeter/0/power 344.69
2024-03-08T16:57:57.504643519Z /em3/emeter/2/power 239.54
2024-03-08T16:57:57.504296802Z /em3/emeter/2/power 239.54
2024-03-08T16:57:57.50423248Z /em3/emeter/1/power 236.73
2024-03-08T16:57:57.504027637Z /em3/emeter/2/power 239.54
2024-03-08T16:57:57.503816909Z /em3/emeter/1/power 236.73
2024-03-08T16:57:57.503769097Z /em3/emeter/0/power 382.78
2024-03-08T16:57:57.503698785Z /em3/emeter/1/power 236.73
2024-03-08T16:57:57.503377589Z /m3/emeter/0/power 382.78

29

You need to go through your mqtt_consumers and figure out first which one of those is sending those messages. Is it the one that is getting passed through the final aggregator?

30

Many thanks to you … i am afraid it has to do with this following problem explained in this thread … and the Final Aggregrator Plug-in is perhaps unable to handle that …
Unable to handle continuous series ….

What do you think?

Best regards
Toni

31

That post is before we added the output_strategy option.

For example, if I send the following messages:

metric,tag=emeter1 value=2
metric,tag=emeter2 value=2
metric,tag=emeter3 value=2

and then within the same aggregation period also send:

metric,tag=emeter1 value=3
metric,tag=emeter2 value=3
metric,tag=emeter3 value=3

and then

metric,tag=emeter1 value=4
metric,tag=emeter2 value=4
metric,tag=emeter3 value=4

I would expect the only metrics to show up would be the final three metrics each with a value of 4.

32

Dear jpowers

First of all, thank you very very much for your valuable help.
What you describe above is exactly what i am looking for …

I do not want to mess up my still smoothly working system.
So i will setup a new dedicated raspi system and then making more tests with
the “Final Aggregrator Plugin” in connection with the shelly 3EM as the only source of mqtt messages on this one.
With this setup we can be sure that no other disturbing source is affecting the mentioned plugin.

I will inform you about the results and expirience made, as soon as possible.

Best regards, Toni

1 Like
33

Hello jpowers

As intended i have built a new system around another raspberry pi 3.

First with the only input from my shelly 3em., and then successivly all the other mqtt inputs from the
original raspi system.
The behavior was as expected, including that of the “Final Aggregarator Plugin”.

But why does it not work on the “old, original” system ?
A lot of Try and error ==> so i changed the name on the original system telegraf.conf file from “grid” to “metrics” ==> and so finally to this measurement only every 30s the shelly 3em metrics are written into the influx db now !!
AS EXPECTED !!

But funny enough, i see that the “grid” measurements is still beeing updated, also after a reboot of the system, and definitely no more reference to “grid” in the actual telegraf.conf file …

Is there a second instance of a telegraf.conf file active ???

Best regards
Toni

34

Excellent, thank you for confirming!

But funny enough, i see that the “grid” measurements is still beeing updated,

Are you running telegraf as a service? Did you restart the service after making the updates?

35

Hi jpowers

Despite not listed as a running service (sudo service --status-all) i tried with the command
“sudo systemctl stop telegraf” to stop a perhaps wrong running service …
This action stopped the update of the “grid” measurement.

Also after several reboots and shutdowns the system is now working correct.

Many thanks once more
Toni

1 Like