fix(dashboard): cross filter chart highlight when filters badge icon clicked

SUMMARY
This PR fixes highlighting the chart emitting a cross filter when magnifier icon in filter's badge is clicked. Before, nothing happened when icon was clicked. Now it highlights and scrolls the chart to view and changes tab if needed.
BEFORE/AFTER SCREENSHOTS OR ANIMATED GIF
Before: see https://github.com/apache/superset/issues/16220
After:
https://user-images.githubusercontent.com/15073128/129217878-f9c4547e-2763-4cc7-b423-cf801d3c4ae6.mov
TESTING INSTRUCTIONS

Enable cross filtering feature flag
Open a dashboard that contains charts that can emit cross filters
Emit cross filter
Click filters badge on another chart and click magnifier icon of the chart that emits the cross filter
Verify that the source chart gets highlighted and scrolled into view

ADDITIONAL INFORMATION

[x] Has associated issue: fixes #16220
[ ] Changes UI
[ ] Includes DB Migration (follow approval process in SIP-59)

[ ] Migration is atomic, supports rollback & is backwards-compatible
[ ] Confirm DB migration upgrade and downgrade tested
[ ] Runtime estimates and downtime expectations provided

[ ] Introduces new feature or API
[ ] Removes existing feature or API

CC @junlincc @jinghua-qa

/testenv up FEATURE_DASHBOARD_NATIVE_FILTERS=true FEATURE_DASHBOARD_CROSS_FILTERS=true

/testenv up FEATURE_DASHBOARD_NATIVE_FILTERS=true FEATURE_DASHBOARD_CROSS_FILTERS=true

🏷 2021.31


Update COVID19_Fallzahlen_Kanton_SG_total.csv

There are two values on 2020-03-04
the second case was announced 2020-03-06
see source
I don't know about time.

@hMutzner good catch!


prometheus 2.22.2 to 2.26.1

In order to leverage more efficient versions of prometheus, this PR bumps from prometheus 2.22.2 to the latest of 2.26.1.
I have tested this on my private environment and haven't experienced problems.
Blobs: if accepted, blobs will need to be stored and uploaded into the blobstore:
wget https://github.com/prometheus/prometheus/releases/download/v2.26.1/prometheus-2.26.1.linux-amd64.tar.gz
bosh add-blob prometheus-2.26.1.linux-amd64.tar.gz prometheus/prometheus-2.26.1.linux-amd64.tar.gz
bosh remove-blob prometheus/prometheus-2.22.2.linux-amd64.tar.gz

This is my first PR, so sorry if I'm wrong.

Hi,
sorry for the late response. There have been some organizational issues on my side. I bumped prometheus to the latest available version, but beside that your PR looked good :)
Thanks for the effort.


[AIRFLOW-282] Remove PR Tool logic that depends on version formatting

This addresses the following issue:
https://issues.apache.org/jira/browse/AIRFLOW-282

[Current coverage][cc-pull] is 64.10%

Merging [#1625][cc-pull] into [master][cc-base-branch] will not change coverage

@@             master      #1625   diff @@
==========================================
  Files           121        121
  Lines          8472       8472
  Methods           0          0
  Messages          0          0
  Branches          0          0
==========================================
  Hits           5431       5431
  Misses         3041       3041
  Partials          0          0

Powered by Codecov. Last updated by [5db7ec7...aa1154d][cc-compare]
[cc-base-branch]: https://codecov.io/gh/apache/incubator-airflow/branch/master?src=pr
[cc-compare]: https://codecov.io/gh/apache/incubator-airflow/compare/5db7ec70f8b45bc4872ae64901d2829efa0e4081...aa1154d764828b1d478eec9cf44d1d1f009e42de
[cc-pull]: https://codecov.io/gh/apache/incubator-airflow/pull/1625?src=pr

LGTM

LGTM


Allow diverted accumulation in calc_network functions.

Adjust accumulate downstream to support diverted flow override.
This function:
https://github.com/USGS-R/nhdplusTools/blob/master/R/calc_network.R#L65
Would be exported and would be able to take a 1:many toID and, if 1:many, require a diverted fraction column in the range 0:100%.

Closing as won't fix here. May tackle in hydroloom down the road.


[slikenet] link failure

Describe the bug
Link slikenet to a program results in a failure.
Environment

OS: Windows 10
Compiler: Visual Studio 2019

To Reproduce
Steps to reproduce the behavior:

./vcpkg install slikenet
#include "slikenet/peer.h"
build

Expected behavior
Successfully linkage.
Failure logs
error LNK2019: unresolved external symbol "public: __thiscall SLNet::BPSTracker::TimeAndValue2::~TimeAndValue2(void)" (??1TimeAndValue2@BPSTracker@SLNet@@QAE@XZ) referenced in function "public: void * __thiscall SLNet::BPSTracker::TimeAndValue2::`vector deleting destructor'(unsigned int)" (??_ETimeAndValue2@BPSTracker@SLNet@@QAEPAXI@Z)

Thanks for posting this issue. After investigation, I found when defining function symbols, slikenet was not export these function symbols.
For example:
struct BPSTracker
{
	BPSTracker();
	~BPSTracker();
	void Reset(const char *file, unsigned int line);
	inline void Push1(CCTimeType time, uint64_t value1) {dataQueue.Push(TimeAndValue2(time,value1),_FILE_AND_LINE_); total1+=value1; lastSec1+=value1;}
//	void Push2(SLNet::TimeUS time, uint64_t value1, uint64_t value2);
	inline uint64_t GetBPS1(CCTimeType time) {(void) time; return lastSec1;}
	inline uint64_t GetBPS1Threadsafe(CCTimeType time) {(void) time; return lastSec1;}
//	uint64_t GetBPS2(RakNetTimeUS time);
//	void GetBPS1And2(RakNetTimeUS time, uint64_t &out1, uint64_t &out2);
	uint64_t GetTotal1(void) const;
//	uint64_t GetTotal2(void) const;

	struct TimeAndValue2
	{
		TimeAndValue2();
		~TimeAndValue2();
		TimeAndValue2(CCTimeType t, uint64_t v1);
	//	TimeAndValue2(SLNet::TimeUS t, uint64_t v1, uint64_t v2);
	//	uint64_t value1, value2;
		uint64_t value1;
		CCTimeType time;
	};

So these function don't support external calls, this is not a bug of VCPKG. You can ask how to use slikenet on the official page of slikenet.
https://github.com/SLikeSoft/SLikeNet
https://www.slikesoft.com/

Thanks for looking into this and sorry that I haven't looked deeper in this before posting the issue. I was sure the dll build and linked correctly, but this was probably with an older version.
I also tested the triplets x64-windows-static-md, x64-windows-static, x86-windows-static-md and x86-windows-static  which works fine.
Maybe it would be advisable to deactivate the triplets x64-windows and x86-windows for now because it is unclear when an official update could happen and I don't know if I could provide a fix.

Thanks for your reply. Since 'slikenet' is a only_static_library, actually when installing 'slikenet:x64-windows' or 'slikenet:x86-windows' the user will get a static library.

Hi,
I have the same problem @Reveares, how I can fix it?
1>Server.obj : error LNK2019: unresolved external symbol "public: struct SLNet::SystemAddress __cdecl SLNet::RakNetSocket2::GetBoundAddress(void)const " (?GetBoundAddress@RakNetSocket2@SLNet@@QEBA?AUSystemAddress@2@XZ) referenced in function main
1>Server.obj : error LNK2019: unresolved external symbol Gets referenced in function main
1>Server.obj : error LNK2001: unresolved external symbol "struct SLNet::SystemAddress const SLNet::UNASSIGNED_SYSTEM_ADDRESS" (?UNASSIGNED_SYSTEM_ADDRESS@SLNet@@3USystemAddress@1@B)
1>Server.obj : error LNK2001: unresolved external symbol "struct SLNet::RakNetGUID const SLNet::UNASSIGNED_RAKNET_GUID" (?UNASSIGNED_RAKNET_GUID@SLNet@@3URakNetGUID@1@B)

Thanks.

There is already an issue, but still unresolved SLikeSoft/SLikeNet#66.

Close this issue until upstream fixes this.


Client and Server only works on same host.

I have the flowjs client on localhost and then hosted the ng-flow server on cloud. Sending the file to the target of the server... Keep getting this error
sha1(): Passing null to parameter #1 ($string) of type string is deprecated"
But when I placed the server and client on same host, it works... but on different host it doesnt work and have added headers such as
header('Content-Type: application/json; charset=UTF-8');
header('Access-Control-Allow-Origin: *');
header('Access-Control-Max-Age: 3600');

For CORS but yet its not working.
On client i have
var flow = new Flow({
  target:'https://my-server-link/upload',
  query:{upload_token:'my_token'},
chunkSize: 1024*1024,
});

Please what is the workable method to use this?

how did you solve the issue ?


Investigate worker connection accept balance

See relevant envoy-dev thread: https://groups.google.com/forum/#!topic/envoy-dev/33QvlXyBinw
Some potential work here:

Performance profile different application loads against Envoy running on different numbers of cores.
Per worker CPU pinning.
Adding EPOLLEXCLUSIVE and EPOLLROUNDROBIN support to libevent and consume them in Envoy
Adding per worker stats for connection load, accept rate, etc.
Investigate Envoy built-in connection rebalance between workers (e.g., allow configurable slew between workers but if some worker gets too far from the highest loaded worker, send a new connection to the other worker).

Note that I've tried variants of ^ over the years and it's extremely difficult to beat the kernel at its own game while generically improving all workloads. With that said, this is a very interesting area to investigate albeit incredibly time consuming.

cc @tonya11en

This can be broken out into a bunch of different tasks, so we can probably have multiple assignees here. I'll volunteer myself to be the first.
An important first step will be to get per-worker stats, so that's where I'll begin working. From there we'll have a basis to begin a more thorough investigation into the performance impact of the items above.

@tonya11en I'm currently working on per-worker stats for event loop duration. Have you started on any of the others? I can take some of these on as well.

@htuch I haven't started work on any of the others yet. Happy to help review.

FYI for those watching I'm chasing an internal performance problem that may be related to this so I will be working on this.

For anyone watching that cares about this issue, I have hacked up code that adds per-worker connection and watchdog miss stats and also re-balances active TCP connections across listeners. I plan on cleaning this up and upstreaming it, but I'm not sure exactly when I will get to it. If anyone urgently wants to work on this let me know. My WIP branch is here: https://github.com/envoyproxy/envoy/tree/cx_rebalance

@mattklein123 would you be able to share the symptoms of internal performance problem that you think are related to this if it is some thing that can be shared? The reason I am asking that is because we are doing some perf tests of a low latency gRPC service and we see that p99 latencies increase as the req/min increase and at the same we also see the p99 connect_time_ms to upstream cluster increases. I was wondering could it be due to high number of worker threads (we left the concurrency arg to default) or could be related to this specific issue? We have n't enabled the dispatcher stats yet. Any thoughts?

@ramaraochavali we are chasing a very similar issue. We have a gRPC service which has many downstream callers (so a large number of incoming mesh connections) and a single upstream callee. At P99 the latency do the upstream service is almost double what the upstream service thinks its downstream latency is.
I've been looking at this for days and I can't find any issue here in terms of blocking or anything other than straight up load, but the delta is pretty large, like 30-40ms in some cases, which is quite large. The only thing I really have right now is this is a straight up load issue where Envoy and the service are context switching back and forth, and Envoy has to deal with a large number of individual connections, and at P99 there are interleavings that just go bad. I'm still looking though.

I forgot to add that balancing the connections and increasing the workers does not seem to improve upstream P99 latency, which is why I think there may be scheduling contention with the service.

@mattklein123 Thanks a lot for the details. This is very useful information. I will share here if we find any thing different from what you have found bases on our tests/tweaks to the config.

@oschaaf see the above thread if you have time. Do we have any NH simulations in which we try to model this type of service mesh use case? The characteristics would be:

gRPC API with a very small request/response payload, so headers/trailers and header compression dominate the workload.
Sidecar pattern with a local service both handling the incoming API as well as making a chained outgoing egress API call.
Lots of incoming callers with persistent connections, but each connection is very low rate, so we get no real benefit of transport level batching.

My suspicion here is that this is a pathological case in which we get a lot of context switching, have to switch/handle a lot of OS connections, etc. I think it's conceivable that there are event loop efficiency issues also but I'm not sure (e.g., the way that we defer writes by activating the write ready event vs. doing them inline).
cc @jmarantz our other perf guru

I have some naive questions :)
I'm inferring from the discussion that the Envoy is sharing a host with its single upstream service process. That right?
How compute-intensive vs i/o-bound is the service? How multi-threaded is the service? I know that Envoy by default will create ~ 1 worker per HW thread; is that how you have it set? If the service had (say) 5 threads could you play games like subtracting 5, or have an Envoy option to use Max(ConfiguredMininumThreadCount, NumHardwareThreads - ConfiguredCarveOut) or something like that?
What happens to perf if you give the upstream a dedicated machine on the same rack as the Envoy?

I'm inferring from the discussion that the Envoy is sharing a host with its single upstream service process. That right?

Envoy is a sidecar to the service (service mesh ingress/ingress)
<Many downstreams> -> <Many downstream egress Envoy> -> <Ingress Envoy> -> <Service> -> <Egress Envoy (same as ingress Envoy)> -> Upstream Service
Running Envoy on more workers does not seem to appreciably improve the situation, and neither does listener connection balancing. I think there are lots of things to try in terms of for example pinning the service to 2 cores and pinning envoy to 2 cores, etc. but I haven't gone there yet. I've mainly been trying to figure out if anything in Envoy is blocking or taking a really long time, and I can't find anything.

@mattklein123 another approach to tackling this kind of hot spotting is to just add a per-TCP connection timeout and do a graceful GOAWAY after this, e.g. every 10 minutes. This will cause the gRPC client to treat this TCP connection as drained and try another one. The new TCP connection will provide better stochastic balancing behavior.

@mattklein123 another approach to tackling this kind of hot spotting is to just add a per-TCP connection timeout and do a graceful GOAWAY after this, e.g. every 10 minutes.

I agree this is a good approach in some cases, but not in every case, particularly the side car case in which the local service may only make a very small number of HTTP/2 connections to the side car. This is why I hacked up "perfect" CX balancing, though it didn't make much of a difference to out workload AFAICT.
The only thing that I still want to spend some time looking at (probably not until next week) is whether how we handle write events is somehow contributing to the problem we are seeing. The current code never writes inline. It always activates an event which then handles the writes.  I don't see anything obvious here but it's worth a look.
I will also work on upstreaming the new stats I added as well as the CX balancing code which I do think is useful for certain cases.

re-balances active TCP connections across listeners

In our case, I doubt we'd need any rebalancing of active connections if the mapping of brand new connections to Envoy threads was done round-robin or least-conns.  In our cases, the many incoming connections have even request load across them, yet most of our Envoy threads appear to have virtually zero connections assigned to them.    Rebalancing active connections would reactively detect and fix this form of imbalance, but there'd still be the initial period of large imbalance we'd have to work around (e.g. by over-scaling or eating bad tail latencies during deployments or scale ups).  In a sense, auto re-balancing further hides this initial connection imbalance.   Thoughts on starting with just better predictability/control over the mapping of new connections to worker threads?

Thoughts on starting with just better predictability/control over the mapping of new connections to worker threads?

Sorry this is what I actually implemented, not rebalancing once accepted. Envoy will likely never support the latter given the architecture.

I think it's conceivable that there are event loop efficiency issues also but I'm not sure (e.g., the way that we defer writes by activating the write ready event vs. doing them inline).

@mattklein123 Is there visibility into event loop queue depth and/or time spent in queue?  Could hidden latency could be while sitting in the event loop queue, e.g. if the processing on that thread was backed up?  Not sure I'm describing this in a way that even makes sense w/ my limited understanding of Envoy :)

@mattklein123 Is there visibility into event loop queue depth and/or time spent in queue?

Not currently, though this would be interesting to track potentially. cc @mergeconflict @jmarantz @oschaaf

Thanks @mattklein123 for the details here. A few points to clarify:
Do you expect the bottleneck originates in the egress path [service -> egress Envoy -> upstream service] or we see the same behavior with direct response from service even when no upstream is involved? If the former, curious if there is a single persistent connection between the upstream callee (service) and its local egress Envoy, or we have a pool of connections that are getting rebalanced across workers with your patch?
In our similar example we needed to leverage a pool of connections which is regularly refreshed every few minutes (as opposed to a single persistent HTTP-2 connection). With your connection (re-)balancing change in place we should no longer have to perform regular connection recycling albeit the need to still maintain a pool of connections from the service -> Envoy to leverage multiple worker threads.
In-line with the point raised by @htuch I think it would be valuable for certain scenarios to expose Envoy cluster options for connection age/time-to-live. In addition to the one mentioned above we have also considered it to rebalance connected Envoy persistent bidi streams to XDS control plane in the mesh, esp on XDS server scale outs.

For those watching this issue, I have a PR up (https://github.com/envoyproxy/envoy/pull/8422) which adds configurable connection balancing for TCP listeners. Please try it out.

FWIW, we started running #8263 in prod and things look pretty imbalanced:
listener.<IP_ADDRESS>_443.downstream_cx_active: 5711
listener.<IP_ADDRESS>_443.downstream_cx_destroy: 895161
listener.<IP_ADDRESS>_443.downstream_cx_total: 895699
listener.<IP_ADDRESS>_443.downstream_pre_cx_active: 0
listener.<IP_ADDRESS>_443.downstream_pre_cx_timeout: 0
listener.<IP_ADDRESS>_443.worker_0.downstream_cx_active: 1
listener.<IP_ADDRESS>_443.worker_0.downstream_cx_total: 221
listener.<IP_ADDRESS>_443.worker_1.downstream_cx_active: 1
listener.<IP_ADDRESS>_443.worker_1.downstream_cx_total: 213
listener.<IP_ADDRESS>_443.worker_10.downstream_cx_active: 5
listener.<IP_ADDRESS>_443.worker_10.downstream_cx_total: 215
listener.<IP_ADDRESS>_443.worker_11.downstream_cx_active: 4
listener.<IP_ADDRESS>_443.worker_11.downstream_cx_total: 237
listener.<IP_ADDRESS>_443.worker_12.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_12.downstream_cx_total: 208
listener.<IP_ADDRESS>_443.worker_13.downstream_cx_active: 6
listener.<IP_ADDRESS>_443.worker_13.downstream_cx_total: 229
listener.<IP_ADDRESS>_443.worker_14.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_14.downstream_cx_total: 212
listener.<IP_ADDRESS>_443.worker_15.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_15.downstream_cx_total: 243
listener.<IP_ADDRESS>_443.worker_16.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_16.downstream_cx_total: 181
listener.<IP_ADDRESS>_443.worker_17.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_17.downstream_cx_total: 216
listener.<IP_ADDRESS>_443.worker_18.downstream_cx_active: 4
listener.<IP_ADDRESS>_443.worker_18.downstream_cx_total: 185
listener.<IP_ADDRESS>_443.worker_19.downstream_cx_active: 7
listener.<IP_ADDRESS>_443.worker_19.downstream_cx_total: 173
listener.<IP_ADDRESS>_443.worker_2.downstream_cx_active: 1
listener.<IP_ADDRESS>_443.worker_2.downstream_cx_total: 235
listener.<IP_ADDRESS>_443.worker_20.downstream_cx_active: 14
listener.<IP_ADDRESS>_443.worker_20.downstream_cx_total: 1105
listener.<IP_ADDRESS>_443.worker_21.downstream_cx_active: 4
listener.<IP_ADDRESS>_443.worker_21.downstream_cx_total: 189
listener.<IP_ADDRESS>_443.worker_22.downstream_cx_active: 4
listener.<IP_ADDRESS>_443.worker_22.downstream_cx_total: 185
listener.<IP_ADDRESS>_443.worker_23.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_23.downstream_cx_total: 166
listener.<IP_ADDRESS>_443.worker_24.downstream_cx_active: 1
listener.<IP_ADDRESS>_443.worker_24.downstream_cx_total: 235
listener.<IP_ADDRESS>_443.worker_25.downstream_cx_active: 49
listener.<IP_ADDRESS>_443.worker_25.downstream_cx_total: 4443
listener.<IP_ADDRESS>_443.worker_26.downstream_cx_active: 6
listener.<IP_ADDRESS>_443.worker_26.downstream_cx_total: 404
listener.<IP_ADDRESS>_443.worker_27.downstream_cx_active: 140
listener.<IP_ADDRESS>_443.worker_27.downstream_cx_total: 12694
listener.<IP_ADDRESS>_443.worker_28.downstream_cx_active: 438
listener.<IP_ADDRESS>_443.worker_28.downstream_cx_total: 60326
listener.<IP_ADDRESS>_443.worker_29.downstream_cx_active: 259
listener.<IP_ADDRESS>_443.worker_29.downstream_cx_total: 33042
listener.<IP_ADDRESS>_443.worker_3.downstream_cx_active: 1
listener.<IP_ADDRESS>_443.worker_3.downstream_cx_total: 212
listener.<IP_ADDRESS>_443.worker_30.downstream_cx_active: 931
listener.<IP_ADDRESS>_443.worker_30.downstream_cx_total: 157107
listener.<IP_ADDRESS>_443.worker_31.downstream_cx_active: 573
listener.<IP_ADDRESS>_443.worker_31.downstream_cx_total: 89898
listener.<IP_ADDRESS>_443.worker_32.downstream_cx_active: 723
listener.<IP_ADDRESS>_443.worker_32.downstream_cx_total: 115034
listener.<IP_ADDRESS>_443.worker_33.downstream_cx_active: 796
listener.<IP_ADDRESS>_443.worker_33.downstream_cx_total: 133812
listener.<IP_ADDRESS>_443.worker_34.downstream_cx_active: 913
listener.<IP_ADDRESS>_443.worker_34.downstream_cx_total: 151900
listener.<IP_ADDRESS>_443.worker_35.downstream_cx_active: 798
listener.<IP_ADDRESS>_443.worker_35.downstream_cx_total: 130669
listener.<IP_ADDRESS>_443.worker_4.downstream_cx_active: 2
listener.<IP_ADDRESS>_443.worker_4.downstream_cx_total: 308
listener.<IP_ADDRESS>_443.worker_5.downstream_cx_active: 8
listener.<IP_ADDRESS>_443.worker_5.downstream_cx_total: 238
listener.<IP_ADDRESS>_443.worker_6.downstream_cx_active: 1
listener.<IP_ADDRESS>_443.worker_6.downstream_cx_total: 250
listener.<IP_ADDRESS>_443.worker_7.downstream_cx_active: 0
listener.<IP_ADDRESS>_443.worker_7.downstream_cx_total: 197
listener.<IP_ADDRESS>_443.worker_8.downstream_cx_active: 6
listener.<IP_ADDRESS>_443.worker_8.downstream_cx_total: 289
listener.<IP_ADDRESS>_443.worker_9.downstream_cx_active: 3
listener.<IP_ADDRESS>_443.worker_9.downstream_cx_total: 228

Are others seeing similar natural imbalances?

Are others seeing similar natural imbalances?

This doesn't surprise me at all. The kernel will generally keep things on the same thread if it thinks that it can avoid context switches, etc. Over the years I have come to realize that in almost all cases the kernel knows what it is doing, but for those small cases where it does not hold true please try https://github.com/envoyproxy/envoy/pull/8422. :)

Have we considered use of SO_REUSEPORT on listeners in order to distribute incoming connections to accept queues via hashing?  Granted, this socket option behavior may be specific to Linux.

Have we considered use of SO_REUSEPORT on listeners in order to distribute incoming connections to accept queues via hashing? Granted, this socket option behavior may be specific to Linux.

Some people already configure SO_REUSEPORT, at least across processes, and there has been discussion about allowing this within the process also so each worker has its own FD. cc @euroelessar. With that said, from previous experience, this still won't fix the case that I just fixed where you have a tiny number of connections and want to make sure they get spread evenly.


any way to improve read_messages speed that is slow?

I wrote a sample program that loops over a directory:

Delete a stream named x
Delete a status stream x-status
Create a stream x
Create a status stream x-status
Append message 1 by 1 to a stream that sends data with S3
Read the status message with read_messages
Delete the file once uploaded

This program loops 5 times. between 30 and 40% of the time is spent on the read_messages function. I tried to read more messages, longer time but it is also long (15 seconds for 37 messages).
This is running on a NUC pc (4 cores I3), it takes around 1 second to delete 1 file, and i will have around 20k to delete every time.
Is there any way to improve the performance?
from email.policy import default
import asyncio
import glob
import pathlib
import time

from stream_manager import (
    ExportDefinition,
    MessageStreamDefinition,
    ReadMessagesOptions,
    ResourceNotFoundException,
    S3ExportTaskDefinition,
    S3ExportTaskExecutorConfig,
    Status,
    StatusConfig,
    StatusLevel,
    StatusMessage,
    StrategyOnFull,
    StreamManagerClient,
    StreamManagerException,
    NotEnoughMessagesException,
)
from stream_manager.util import Util

if __name__ == "__main__":
    stream_name = "s3-upload"
    s3_bucket = "my-bucket"
    sm_host = "<IP_ADDRESS>"
    sm_port = 8088
    client = StreamManagerClient(host=sm_host, port=sm_port)
    tmp_path = "/greengrass/xxx/"
    status_stream_name = f"{stream_name}-status"

    main_counter = 0

    while True:
        try:
            # Try deleting the status stream (if it exists) so that we have a fresh start
            try:
                client.delete_message_stream(stream_name=status_stream_name)
            except ResourceNotFoundException:
                pass

            # Try deleting the stream (if it exists) so that we have a fresh start
            try:
                client.delete_message_stream(stream_name=stream_name)
            except ResourceNotFoundException:
                pass

            exports = ExportDefinition(
                s3_task_executor=[
                    S3ExportTaskExecutorConfig(
                        identifier=f"S3TaskExecutor-{stream_name}",
                        status_config=StatusConfig(
                            status_level=StatusLevel.INFO,  # Default is debug level statuses.
                            status_stream_name=status_stream_name,
                        ),
                    )
                ]
            )

            # Create the Status Stream.
            client.create_message_stream(
                MessageStreamDefinition(
                    name=status_stream_name,
                    strategy_on_full=StrategyOnFull.OverwriteOldestData,
                )
            )

            # Create the message stream with the S3 Export definition.
            client.create_message_stream(
                MessageStreamDefinition(
                    name=stream_name,
                    strategy_on_full=StrategyOnFull.OverwriteOldestData,
                    export_definition=exports,
                )
            )
        except asyncio.TimeoutError as e:
            print(f"Timed out while executing: {e}")
        except Exception as e:
            print(f"Exception while running: {e}")
        try:
            files_found = glob.glob(f"{tmp_path}/**", recursive=True)
            count = 0
            for wr_file in files_found:
                if pathlib.Path(wr_file).is_file():
                    s3_export_task_definition = S3ExportTaskDefinition(
                        input_url=f"file://{wr_file}",
                        bucket=s3_bucket,
                        key=wr_file.split(tmp_path)[1],
                    )
                    print(wr_file)
                    sequence_nbr = client.append_message(
                        stream_name,
                        Util.validate_and_serialize_to_json_bytes(
                            s3_export_task_definition
                        ),
                    )
                    count += 1
                    if count == 1:
                        break
            print(
                f"Successfully appended all files ({sequence_nbr}) ",
            )
        except asyncio.TimeoutError as e:
            print(f"Timed out while executing: {e}")
        except Exception as e:
            print(f"Exception while running: {e}")

        try:
            all_messages_sent = False
            while not all_messages_sent:
                try:
                    start_time = time.time()
                    messages_list = client.read_messages(
                        status_stream_name,
                        ReadMessagesOptions(
                            min_message_count=1
                        ),
                    )
                    end_time = time.time()
                    print(
                        f"[{end_time - start_time}] Got {len(messages_list)} messages from stream"
                    )

                    failure_flag = False
                    for index, message in enumerate(messages_list):
                        # Deserialize the status message first.
                        status_message = Util.deserialize_json_bytes_to_obj(
                            message.payload, StatusMessage
                        )
                        local_path = status_message.status_context.s3_export_task_definition.input_url.split(
                            "file://"
                        )[
                            1
                        ]
                        s3_path = f"s3://{status_message.status_context.s3_export_task_definition.bucket}/{status_message.status_context.s3_export_task_definition.key}"

                        if (
                            pathlib.Path(local_path).is_file()
                            and status_message.status == Status.Success
                        ):
                            print(
                                f"[{index+1}/{len(messages_list)}] Successfully uploaded file at path {local_path} to S3: "
                                f"{s3_path}."
                            )
                            # os.unlink(local_path) # We normally delete the file here
                        elif status_message.status == Status.InProgress:
                            print(
                                f"[{index+1}/{len(messages_list)}] File upload is in progress."
                            )
                            failure_flag = True
                        elif status_message.status in [
                            Status.Failure,
                            Status.Canceled,
                        ]:
                            print(
                                f"[{index+1}/{len(messages_list)}] Unable to upload file at path {local_path} to S3."
                                f"Message: {status_message.message}",
                            )
                            failure_flag = True
                    all_messages_sent = not failure_flag
                except NotEnoughMessagesException as e:
                    all_messages_sent = False
        except StreamManagerException as e:
            print(f"Exception while running: {e}")
        main_counter += 1
        if main_counter == 5:
            break

Hi @Minipada,
The status stream should only have messages in it when the file upload succeeds or fails, so you aren't actually trying to optimize the speed of read_message, right? What you really are looking for is faster uploading it would seem to me. Does that make sense?
I'd suggest that you append more than one S3 upload task so that work can be batched a bit more. Ultimately you are going to be limited by the disk read and network upload bandwidth since you are uploading files to S3.

Sounds like there's no issue here so I'm going to close out this issue. Reopen if there's something else to discuss on this issue.

Hi @MikeDombo , thanks for closing the issue. I have tested it more today, on 5000 more files and it worked well. Thanks!


NPM 2.3.0 package logs to the console

Not sure where to ask this, but I've installed 2.3.0 from NPM, and the following gets logged out to the console:
console.log('should rendre as tablist?', this.renderAsTablist);

https://www.npmjs.com/package/vue-ssr-carousel?activeTab=code (line 676)
I've had to revert to 2.2.0.
It's not from the source, but can this be removed, and repackaged?
Thanks for the good work.

Thanks for the reminder, 2.3.1 is published to npm now


funds-manager: Add internal wallet id mapping & use internal transaction

Purpose
This PR makes the following changes to bring the funds-manager into a working state with our testnet environment:

Remove the sandbox flag on the Fireblocks client
Add an internal_wallet_id mapping to the hot wallets table. Fireblocks requires that we specify whitelisted withdrawals to a wallet ID rather than its address. I manually added wallet IDs for each of the existing hot wallets
Use a create_transaction_peer with PeerType::INTERNAL_WALLET to withdraw from Fireblocks. This allows us to control where our funds flow to via a TAP

Testing

Tested withdrawing from fireblocks and depositing into it

LGTM, does this need a migration due to the new col?

forgot to commit it, thanks


Auto disarming charms and cheese, don't get recorded for the hunt.

for example sunrise cheese

for now all hunts without cheese don't get recorded