Questions tagged with AWS IoT Core

Content language: English

Sort by most recent

Browse through the questions and answers listed below or filter and sort to narrow down your results.

AWS Greengrass doesn't send data to AWS Kinesis

The main purpose of my program is to connect to an incoming MQTT channel, and send the data received to my AWS Kinesis Stream called "MyKinesisStream". Here is my code: ``` import argparse import logging import random from paho.mqtt import client as mqtt_client from stream_manager import ( ExportDefinition, KinesisConfig, MessageStreamDefinition, ResourceNotFoundException, StrategyOnFull, StreamManagerClient, ReadMessagesOptions, ) broker = 'localhost' port = 1883 topic = "clients/test/hello/world" client_id = f'python-mqtt-{random.randint(0, 100)}' username = '...' password = '...' logging.basicConfig(level=logging.INFO) logger = logging.getLogger() args = "" def connect_mqtt() -> mqtt_client: def on_connect(client, userdata, flags, rc): if rc == 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d\n", rc) client = mqtt_client.Client(client_id) client.username_pw_set(username, password) client.on_connect = on_connect client.connect(broker, port) return client def sendDataToKinesis( stream_name: str, kinesis_stream_name: str, payload, batch_size: int = None, ): try: print("Debug: sendDataToKinesis with params:", stream_name + " | ", kinesis_stream_name, " | ", batch_size) print("payload:", payload) print("type payload:", type(payload)) except Exception as e: print("Error while printing out the parameters", str(e)) logger.exception(e) try: # Create a client for the StreamManager kinesis_client = StreamManagerClient() # Try deleting the stream (if it exists) so that we have a fresh start try: kinesis_client.delete_message_stream(stream_name=stream_name) except ResourceNotFoundException: pass exports = ExportDefinition( kinesis=[KinesisConfig( identifier="KinesisExport" + stream_name, kinesis_stream_name=kinesis_stream_name, batch_size=batch_size, )] ) kinesis_client.create_message_stream( MessageStreamDefinition( name=stream_name, strategy_on_full=StrategyOnFull.OverwriteOldestData, export_definition=exports ) ) sequence_no = kinesis_client.append_message(stream_name=stream_name, data=payload) print( "Successfully appended message to stream with sequence number ", sequence_no ) readValue = kinesis_client.read_messages(stream_name, ReadMessagesOptions(min_message_count=1, read_timeout_millis=1000)) print("DEBUG read test: ", readValue) except Exception as e: print("Exception while running: " + str(e)) logger.exception(e) finally: # Always close the client to avoid resource leaks print("closing connection") if kinesis_client: kinesis_client.close() def subscribe(client: mqtt_client, args): def on_message(client, userdata, msg): print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic") sendDataToKinesis(args.greengrass_stream, args.kinesis_stream, msg.payload, args.batch_size) client.subscribe(topic) client.on_message = on_message def run(args): mqtt_client_instance = connect_mqtt() subscribe(mqtt_client_instance, args) mqtt_client_instance.loop_forever() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument('--greengrass-stream', required=False, default='...') parser.add_argument('--kinesis-stream', required=False, default='MyKinesisStream') parser.add_argument('--batch-size', required=False, type=int, default=500) return parser.parse_args() if __name__ == '__main__': args = parse_args() run(args) ``` (the dotted parts ... are commented out as they are sensitive information, but they are correct values.) The problem is that it just won't send any data to our kinesis stream. I get the following STDOUT from the run: ``` 2022-11-25T12:13:47.640Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. Connected to MQTT Broker!. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. Received `{"machineId":2, .... "timestamp":"2022-10-24T12:21:34.8777249Z","value":true}` from `clients/test/hello/world` topic. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. Debug: sendDataToKinesis with params: test | MyKinesisStream | 100. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. payload: b'{"machineId":2,... ,"timestamp":"2022-10-24T12:21:34.8777249Z","value":true}'. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. type payload: <class 'bytes'>. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. Successfully appended message to stream with sequence number 0. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. DEBUG read test: [<Class Message. stream_name: 'test', sequence_number: 0, ingest_time: 1669376980985, payload: b'{"machineId":2,"mach'>]. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} 2022-11-25T12:13:47.641Z [INFO] (Copier) jp.co.xyz.StreamManagerKinesis: stdout. closing connection. {scriptName=services.jp.co.xyz.StreamManagerKinesis.lifecycle.Run, serviceName=jp.co.xyz.StreamManagerKinesis, currentState=RUNNING} ``` So we can see that the data arrives from MQTT, the python code executes the append message, and it seems that my kinesis streams have the information as it can read it in the next step... then closes the connection without any error. But the problem is, that from AWS side, we cannot see the data arriving on the stream: [![screnshot of the aws console][1]][1] What can be the problem here? Our greengrass core is configured properly, can be accessed from the AWS, and the Component is running and healthy also: [![Screenshot of IoT Core status][2]][2] [![Screenshot of the state if the StreamManager component][3]][3] [1]: https://i.stack.imgur.com/wN5I4.png [2]: https://i.stack.imgur.com/JMJmn.png [3]: https://i.stack.imgur.com/2qnfn.png
0
answers
0
votes
8
views
ForestG
asked a day ago

IDT 4.5.11 Error in retrieving AwsCredentials from TES

I am using IDT 4.5.11 running test suite GGV2Q_2.4.1 on Greengrass 2.7.0. I failed the [mqttpubsub] and [cloudcomponent] test groups. Below are logs from mqttpubsub/greengrass.log, I am not sure what credential is expired as the Access & Secret Key are both newly created. Any idea how to solve this issue? 2022-11-21T19:04:13.050Z [DEBUG] (pool-2-thread-6) com.aws.greengrass.tes.CredentialRequestHandler: Received response from cloud: response code 200, not logging credentials. {iotCredentialsPath=/role-aliases/idt-a1a2f5615374392fcbc9-ggc-role-alias/credentials, statusCode=200} 2022-11-21T19:04:13.050Z [ERROR] (pool-2-thread-6) com.aws.greengrass.tes.CredentialRequestHandler: Unable to cache expired credentials which expired at 2022-11-21T12:03:44Z. {iotCredentialsPath=/role-aliases/idt-a1a2f5615374392fcbc9-ggc-role-alias/credentials} 2022-11-21T19:04:13.051Z [ERROR] (pool-2-thread-6) com.aws.greengrass.tes.CredentialRequestHandler: Error in retrieving AwsCredentials from TES. {iotCredentialsPath=/role-aliases/idt-a1a2f5615374392fcbc9-ggc-role-alias/credentials, credentialData=TES responded with expired credentials: {"credentials":{"accessKeyId":"ASIAZ625MVFNBTBLFT6Z","secretAccessKey":"WD5VbRjGmp3mDhqSL6udPjID9cZ+b2WEQ4Rf/CWu","sessionToken":"IQoJb3JpZ2luX2VjEIP//////////wEaCXVzLXdlc3QtMiJIMEYCIQCJXJM5KGN71gUQvo5F5j7T2n74d4qc+i066ZDrDIYOpgIhAPJG1U4LyhLFkOjQQbCo9Bq1ht9boqtfiQUE2I03P/OaKqwECIz//////////wEQABoMNjg0NzA2Mjc3NzIyIgzH/z2OTB04Urnlr3wqgARdz+wbs2oS/Hr1h2Ep8KCHqbBaixaDRDCa3HawQkI+UW3r+A7JcXiiM2MZcNNjgr2X+YOdgwpPA6Pd+3b7V8fXcQudkxQfSCXJcoB0gssCi/IrlNe8mdRViN/hfxNIk1H91RxygSE/epH+f2WEfh7lt81ZUNTelBfxor36InjawZu3t0ZLv74ZIlJBcjxNSajKaxg8lKB+sxoQzaF6OlR0nrkx2W0Np9ikFk1eYeWCWyOgnQ8PhcRKivTU77b7KXlOCCkkpITxMU0/gf8EuWfsy9Ik9KQZjr/CNxgvIo8nE7VE0Be1TEvRdRwP5VsH3+EAxCv5VJkXFg8vka7R01JKkweDXEb3JVJhcKz82xr9hIFo5an34InDbc3ssC/I19/ZSZHygVE+aEjgujpmrhLkDUThSsjm8E6+rHdDoY3W+ugL2ik/Yjwk04jT/nl49T86woRk+bU08VW03Vwk+kfqpMX34mP4FntuChc8Niig3E0jWznlX4N595tg+LUj6LXjpaYFNkQbena1b7iXalA+zxHw11GS+V5oisipH9zsCNrdI9hhzqpPt+YhyPCLg28H7hXNLUezxdJAaVjlAuOS8SUR9R35A+POFuUp5T0nLi5XutGPq7r2q+kh1RXCbLKlCM4ajyejb6jsr4tlyo/VUPY/FVXghoIVaPDOBUHh3zCQtu2bBjrFAcsqM7H6gspDrVero0uPfAuHgExojSa3UsIu6kgEWmPJeFTonceeZemuQp8flqxLsdnRcC5IAHd19XA3w35MWGZ+9tcSOVYSpzYepdXw31iVMoFqs1bYLF5NtbqlTZKHrM1O7M5rZsyH7Z2D4JzMMqAwGUhb4bdpp+1Aouvlpm6NE63FPqrvsh0SNpGurRYE2Ijwtj0Kmt1Q+oM0uxrh+gZB77aWGxzjVnQORpoVGWn6oynNxR/LlI5LfmGfDawKMozMQxpk","expiration":"2022-11-21T12:03:44Z"}}} 2022-11-21T19:04:13.051Z [DEBUG] (pool-2-thread-6) com.aws.greengrass.componentmanager.builtins.S3Downloader: get-bucket-location. task failed and will be retried. {task-attempt=7, componentIdentifier=aws.greengrass.IotMqttSubscriber, artifactUri=s3://idt-a1a2f5615374392fcbc9-gg-component-store/greengrass/components/artifacts/aws-greengrass-testing-features-mqtt.zip} software.amazon.awssdk.core.exception.SdkClientException: Failed to fetch credentials at software.amazon.awssdk.core.exception.SdkClientException$BuilderImpl.build(SdkClientException.java:98) at software.amazon.awssdk.core.exception.SdkClientException.create(SdkClientException.java:39) at com.aws.greengrass.tes.LazyCredentialProvider.resolveCredentials(LazyCredentialProvider.java:31) at software.amazon.awssdk.awscore.internal.AwsExecutionContextBuilder.resolveCredentials(AwsExecutionContextBuilder.java:171) at software.amazon.awssdk.awscore.internal.AwsExecutionContextBuilder.invokeInterceptorsAndCreateExecutionContext(AwsExecutionContextBuilder.java:108) at software.amazon.awssdk.awscore.client.handler.AwsSyncClientHandler.invokeInterceptorsAndCreateExecutionContext(AwsSyncClientHandler.java:69) at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.lambda$execute$1(BaseSyncClientHandler.java:78) at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.measureApiCallSuccess(BaseSyncClientHandler.java:175) at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.execute(BaseSyncClientHandler.java:76) at software.amazon.awssdk.core.client.handler.SdkSyncClientHandler.execute(SdkSyncClientHandler.java:45) at software.amazon.awssdk.awscore.client.handler.AwsSyncClientHandler.execute(AwsSyncClientHandler.java:56) at software.amazon.awssdk.services.s3.DefaultS3Client.getBucketLocation(DefaultS3Client.java:3382) at com.aws.greengrass.componentmanager.builtins.S3Downloader.lambda$getRegionClientForBucket$2(S3Downloader.java:134) at com.aws.greengrass.util.RetryUtils.runWithRetry(RetryUtils.java:50) at com.aws.greengrass.componentmanager.builtins.S3Downloader.getRegionClientForBucket(S3Downloader.java:133) at com.aws.greengrass.componentmanager.builtins.S3Downloader.getDownloadSize(S3Downloader.java:115) at com.aws.greengrass.componentmanager.ComponentManager.prepareArtifacts(ComponentManager.java:420) at com.aws.greengrass.componentmanager.ComponentManager.preparePackage(ComponentManager.java:377) at com.aws.greengrass.componentmanager.ComponentManager.lambda$preparePackages$1(ComponentManager.java:338) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829)
2
answers
1
votes
38
views
asked 5 days ago