diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md index e339db6..2d8968c 100644 --- a/.github/ISSUE_TEMPLATE/bug_report.md +++ b/.github/ISSUE_TEMPLATE/bug_report.md @@ -1,12 +1,12 @@ --- name: Bug Report -about: Report a bug in cloud-runtimes-jvm +about: Report a bug in the Capa Java AWS adapters title: '' labels: kind/bug assignees: '' --- -- [ ] I have searched the [issues](https://github.com/reactivegroup/capa-aws/issues) of this repository and believe that this is not a duplicate. +- [ ] I have searched the [issues](https://github.com/capa-cloud/capa-java-aws/issues) of this repository and believe that this is not a duplicate. ### Environment @@ -34,8 +34,8 @@ assignees: '' - - - + + + -RELEASE NOTE: \ No newline at end of file +RELEASE NOTE: diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 6c0af76..0dce75c 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -1,126 +1,36 @@ -# Define the workflow's name -name: "Build and Test" -# Triggered after push, pull_request and workflow_dispatch events -on: [push, pull_request, workflow_dispatch] +name: Build and Test -# Define the environment variables -env: - FAIL_FAST: 0 - SHOW_ERROR_DETAIL: 1 +on: + push: + pull_request: + workflow_dispatch: -jobs: - build-source: - name: "Build Source Code" - runs-on: ubuntu-latest - env: - # The default JDK version - JDK_VER: 8 - outputs: - version: ${{ steps.capa-aws-version.outputs.version }} - steps: - - name: "Checkout the source code" - uses: actions/checkout@v2 - with: - path: capa-aws - - name: "Set up OpenJDK ${{ env.JDK_VER }}" - uses: actions/setup-java@v1 - with: - java-version: ${{ env.JDK_VER }} - - name: "Cache local Maven repository" - uses: actions/cache@v2 - with: - path: ~/.m2/repository - key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} - - name: "Cache cloud runtimes jvm" - uses: actions/cache@v2 - with: - path: ~/.m2/repository/group/rxcloud/capa - key: ${{ runner.os }}-capa-aws-snapshot-${{ github.sha }} - - name: "Build cloud runtimes jvm" - run: | - cd ${{ github.workspace }}/capa-aws - mvn clean - mvn --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean source:jar install -Pjacoco,rat,checkstyle -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Dmaven.wagon.http.retryHandler.count=5 -Dmaven.test.skip=true -Dmaven.test.skip.exec=true -Dgpg.skip=true - - name: "Pack rat file if failure" - if: failure() - run: 7z a ${{ github.workspace }}/rat.zip *rat.txt -r - - name: "Upload rat file if failure" - if: failure() - uses: actions/upload-artifact@v2 - with: - name: "rat-file" - path: ${{ github.workspace }}/rat.zip - - name: "Pack checkstyle file if failure" - if: failure() - run: 7z a ${{ github.workspace }}/checkstyle.zip *checkstyle* -r - - name: "Upload checkstyle file if failure" - if: failure() - uses: actions/upload-artifact@v2 - with: - name: "checkstyle-file" - path: ${{ github.workspace }}/checkstyle.zip +permissions: + contents: read - coverage: - name: "Code Coverage" - needs: [ build-source ] +jobs: + test: + name: Java ${{ matrix.java }} runs-on: ubuntu-latest - steps: - - name: "Checkout the source code" - uses: actions/checkout@v2 - with: - path: capa-aws - - name: "Set up OpenJDK 8" - uses: actions/setup-java@v1 - with: - distribution: 'adopt' - java-version: 8 - - uses: actions/cache@v2 - name: "Cache local Maven repository" - with: - path: ~/.m2/repository - key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} - restore-keys: | - ${{ runner.os }}-maven- - - name: "Calculate code coverage" - run: | - cd ${{ github.workspace }}/capa-aws - mvn --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean test verify -Pjacoco -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Dmaven.wagon.http.retryHandler.count=5 -DskipTests=false -DskipIntegrationTests=false -Dcheckstyle.skip=true -Drat.skip=true -Dmaven.javadoc.skip=true -Dgpg.skip=true - - name: "Upload to Codecov" - uses: codecov/codecov-action@v1 - with: - token: ${{ secrets.CODECOV_TOKEN }} - file: ./**/target/site/jacoco/jacoco.xml - name: codecov - - unit-test: - needs: [ build-source ] - name: "Unit Test On ${{ matrix.os }} (OpenJDK: ${{ matrix.jdk }})" - runs-on: ${{ matrix.os }} strategy: fail-fast: false matrix: - os: [ ubuntu-latest ] - jdk: [ 8, 11 ] - env: - DISABLE_FILE_SYSTEM_TEST: true + java: ["8", "11"] steps: - - name: "Checkout the source code" - uses: actions/checkout@v2 - with: - path: capa-aws - - name: "Set up OpenJDK ${{ matrix.jdk }}" - uses: actions/setup-java@v1 - with: - java-version: ${{ matrix.jdk }} - - uses: actions/cache@v2 - name: "Cache local Maven repository" - with: - path: ~/.m2/repository - key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }} - restore-keys: | - ${{ runner.os }}-maven- - - name: "Unit Test" - timeout-minutes: 10 - run: | - cd ${{ github.workspace }}/capa-aws - mvn --batch-mode --no-snapshot-updates -e --no-transfer-progress --fail-fast clean test verify -Pjacoco,rat,checkstyle -Dmaven.wagon.httpconnectionManager.ttlSeconds=120 -Dmaven.wagon.http.retryHandler.count=5 -DskipTests=false -Dcheckstyle.skip=false -Drat.skip=false -Dmaven.javadoc.skip=true -Dgpg.skip=true + - name: Check out source + uses: actions/checkout@v7 + - name: Set up Java ${{ matrix.java }} + uses: actions/setup-java@v5 + with: + distribution: temurin + java-version: ${{ matrix.java }} + cache: maven + - name: Verify + run: >- + mvn --batch-mode --no-transfer-progress --fail-fast clean verify + -Pjacoco,rat,checkstyle + -DskipTests=false + -Dcheckstyle.skip=false + -Drat.skip=false + -Dmaven.javadoc.skip=true + -Dgpg.skip=true diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..7f129f1 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,7 @@ +# Repository Guidelines + +- GitHub: `capa-cloud/capa-java-aws` +- Category: Java AWS adapter/runtime project. +- Public documentation: `https://capa.rxcloud.group/` + +Keep cloud credentials out of source. Validate adapter changes with local tests or mocked AWS clients before documenting them as ready. diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..261eeb9 --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed 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. diff --git a/README.md b/README.md index f2ba091..0c6d976 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,128 @@ -# Capa Aws +

+ Capa AWS +

-Cloud Application Api implement by AWS. +# Capa Java AWS Adapters -## Features +AWS SPI modules for the [Capa Java SDK](https://github.com/capa-cloud/capa-java). Each module connects a specific Capa component surface to AWS SDK for Java v2 services. -### RPC +[Documentation](https://capa.rxcloud.group/) · [Issues](https://github.com/capa-cloud/capa-java-aws/issues) -AWS App Mes +> This repository contains adapters, not a standalone application. Applications must include `capa-sdk` and only the adapter modules they use. -### Configuration +## Requirements and versions -AWS App Config +- Java 8 or 11 +- Maven 3.8.1 or later +- Capa Java `1.11.13.2.RELEASE` +- Capa AWS adapters `1.11.13.5.RELEASE` +- AWS credentials and IAM permissions for the selected services -### Telemetry +There is no `group.rxcloud:capa-spi-aws:1.11.13.5.RELEASE` aggregate artifact. Depend on the individual modules below. -AWS Cloud Watch +## Modules + +| Artifact | AWS integration | Implemented surface | +| --- | --- | --- | +| `capa-spi-aws-mesh` | App Mesh-compatible service addressing | `AwsCapaHttp`, RPC options and serializer loading | +| `capa-spi-aws-config` | AWS AppConfig | `AwsCapaConfigStore`, polling and serialization | +| `capa-spi-aws-telemetry` | Amazon CloudWatch | Metric export and AWS trace-context propagation | +| `capa-spi-aws-log` | Amazon CloudWatch Logs | Log4j/Logback appenders and log delivery | +| `capa-spi-aws-infrastructure` | Shared Capa infrastructure | Common environment integration used by the other modules | + +The table describes code present in this repository. It does not claim that AWS provisions service discovery, routing, mTLS, alarms, or X-Ray resources for the application. Configure those capabilities separately in AWS when required. + +## Add an adapter + +Choose only the modules required by the application. For example: + +```xml + + 1.11.13.2.RELEASE + 1.11.13.5.RELEASE + + + + + group.rxcloud + capa-sdk + ${capa.version} + + + + group.rxcloud + capa-spi-aws-mesh + ${capa.aws.version} + runtime + + + + group.rxcloud + capa-spi-aws-config + ${capa.aws.version} + runtime + + +``` + +Use `capa-spi-aws-telemetry` and `capa-spi-aws-log` in the same way when those integrations are needed. + +Adapter classes are discovered through Capa component resource mappings. Do not instantiate internal SPI classes as application clients. The relevant defaults and mapping files are: + +- [Mesh RPC settings](capa-spi-aws-mesh/src/main/resources/capa-component-rpc-aws.properties) +- [AppConfig settings](capa-spi-aws-config/src/main/resources/capa-component-configuration-aws.properties) +- [Telemetry settings](capa-spi-aws-telemetry/src/main/resources/capa-component-telemetry-aws.properties) +- [Log component mapping](capa-spi-aws-log/src/main/resources/capa-component-log.properties) + +Copy environment-specific values into the application's configuration. Do not modify and publish credentials in these resource files. + +## AWS credentials + +AWS clients use the AWS SDK default credential and Region resolution behavior. Prefer short-lived credentials supplied by the workload environment, such as an ECS task role or EC2 instance profile. Local development can use environment variables or an AWS shared configuration profile. + +Never commit `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, session tokens, account identifiers, or private endpoints to this repository. + +## Repository layout + +```text +. +├── capa-spi-aws-mesh/ +├── capa-spi-aws-config/ +├── capa-spi-aws-telemetry/ +├── capa-spi-aws-log/ +├── capa-spi-aws-infrastructure/ +├── example/ # Current logging example +└── pom.xml +``` + +The current [`example/`](example/) module demonstrates the logging integration. Treat module tests as implementation examples for configuration, mesh, and telemetry until dedicated runnable samples are added. + +## Build and verify + +```bash +git clone https://github.com/capa-cloud/capa-java-aws.git +cd capa-java-aws +mvn --batch-mode --no-transfer-progress --fail-fast clean verify \ + -Pjacoco,rat,checkstyle \ + -DskipTests=false \ + -Dcheckstyle.skip=false \ + -Drat.skip=false \ + -Dmaven.javadoc.skip=true \ + -Dgpg.skip=true +``` + +CI runs this command on Java 8 and Java 11. Tests mock or isolate AWS calls where possible; production readiness still requires integration tests with the target IAM policies, Region, network, and AWS resources. + +## Contributing + +1. Create a branch from `master`. +2. Keep adapter behavior compatible with Capa Java `1.11.13.2.RELEASE`. +3. Add mocked tests for adapter behavior and integration tests for changes that depend on AWS responses. +4. Update resource examples and this README when configuration keys or artifacts change. +5. Run the full verification command above before opening a pull request. + +The repository enforces Apache RAT and Checkstyle. It does not currently configure Google Java Format or SpotBugs. + +## License + +Apache License 2.0. See [LICENSE](LICENSE). diff --git a/capa-spi-aws-config/pom.xml b/capa-spi-aws-config/pom.xml index 96f4180..5338fef 100644 --- a/capa-spi-aws-config/pom.xml +++ b/capa-spi-aws-config/pom.xml @@ -23,7 +23,7 @@ capa-aws-parent group.rxcloud - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE capa-spi-aws-config diff --git a/capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java b/capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java index a45c531..5c56d3e 100644 --- a/capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java +++ b/capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java @@ -17,12 +17,11 @@ package group.rxcloud.capa.spi.aws.config; import com.google.common.collect.Lists; -import group.rxcloud.capa.addons.foundation.CapaFoundation; -import group.rxcloud.capa.addons.foundation.FoundationType; import group.rxcloud.capa.component.CapaConfigurationProperties; import group.rxcloud.capa.component.configstore.ConfigurationItem; import group.rxcloud.capa.component.configstore.StoreConfig; import group.rxcloud.capa.component.configstore.SubscribeResp; +import group.rxcloud.capa.infrastructure.CapaEnvironment; import group.rxcloud.capa.infrastructure.exceptions.CapaErrorContext; import group.rxcloud.capa.infrastructure.exceptions.CapaException; import group.rxcloud.capa.infrastructure.serializer.CapaObjectSerializer; @@ -133,7 +132,7 @@ protected Mono>> doGet(String appId, String group, return Mono.error(new CapaException(CapaErrorContext.PARAMETER_ERROR, "keys is null or empty")); } - String applicationName = String.format(APPCONFIG_NAME_FORMAT, appId, CapaFoundation.getEnv(FoundationType.TRIP)); + String applicationName = String.format(APPCONFIG_NAME_FORMAT, appId, CapaEnvironment.Provider.getInstance().getDeployEnv()); String configurationName = keys.get(0); //init config and create subscribe polling @@ -147,7 +146,7 @@ protected Mono>> doGet(String appId, String group, @Override protected Flux> doSubscribe(String appId, String group, String label, List keys, Map metadata, TypeRef type) { - String applicationName = String.format(APPCONFIG_NAME_FORMAT, appId, CapaFoundation.getEnv(FoundationType.TRIP)); + String applicationName = String.format(APPCONFIG_NAME_FORMAT, appId, CapaEnvironment.Provider.getInstance().getDeployEnv()); String configurationName = keys.get(0); initAndSubscribe(applicationName, configurationName, group, label, metadata, type); @@ -184,10 +183,10 @@ private synchronized Configuration initConfig(String applicationName, Str LOGGER.debug("[[type=Capa.Config.initConfig]] call getconfiguration in init process,request:{}", request); return Mono.fromCallable(() -> { - GetConfigurationResponse response = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS); - LOGGER.debug("[[type=Capa.Config.initConfig]] call getconfiguration in init process,response:{}", response); - return response; - }) + GetConfigurationResponse response = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS); + LOGGER.debug("[[type=Capa.Config.initConfig]] call getconfiguration in init process,response:{}", response); + return response; + }) .doOnError(e -> LOGGER.warn("[[type=Capa.Config.initConfig]] error occurs when getconfiguration in init process, request:{}", request, e)) .map(resp -> initConfigurationItem(applicationName, configurationName, type, resp.content(), resp.configurationVersion())) .block(); @@ -207,39 +206,39 @@ private synchronized void createSubscribePolling(String applicationName, Str return; } Flux.create(fluxSink -> { - AwsCapaConfigurationScheduler.INSTANCE.configSubscribePollingScheduler - .schedulePeriodically(() -> { - - // update subscribed status if needs - getConfiguration(applicationName, configurationName).getSubscribed().compareAndSet(false, true); - - String version = getCurVersion(applicationName, configurationName); - - GetConfigurationRequest request = GetConfigurationRequest.builder() - .application(applicationName) - .clientId(UUID.randomUUID().toString()) - .configuration(configurationName) - .clientConfigurationVersion(version) - .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getConfigAwsAppConfigEnv()) - .build(); - LOGGER.debug("[[type=Capa.Config.subscribePolling]] subscribe polling task start,request:{}", request); - - GetConfigurationResponse resp = null; - try { - resp = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - //catch error,log error and not trigger listeners - LOGGER.warn("[[type=Capa.Config.subscribePolling]] error occurs when getConfiguration in polling process,configurationName:{},version:{}", request.configuration(), request.clientConfigurationVersion(), e); - } - - LOGGER.debug("[[type=Capa.Config.subscribePolling]] subscribe polling task end,response:{}", resp); - - if (resp != null && !Objects.equals(resp.configurationVersion(), version)) { - fluxSink.next(resp); - } - // todo: make the polling frequency configurable - }, 0, 5, TimeUnit.SECONDS); - }) + AwsCapaConfigurationScheduler.INSTANCE.configSubscribePollingScheduler + .schedulePeriodically(() -> { + + // update subscribed status if needs + getConfiguration(applicationName, configurationName).getSubscribed().compareAndSet(false, true); + + String version = getCurVersion(applicationName, configurationName); + + GetConfigurationRequest request = GetConfigurationRequest.builder() + .application(applicationName) + .clientId(UUID.randomUUID().toString()) + .configuration(configurationName) + .clientConfigurationVersion(version) + .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getConfigAwsAppConfigEnv()) + .build(); + LOGGER.debug("[[type=Capa.Config.subscribePolling]] subscribe polling task start,request:{}", request); + + GetConfigurationResponse resp = null; + try { + resp = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException | ExecutionException | TimeoutException e) { + //catch error,log error and not trigger listeners + LOGGER.warn("[[type=Capa.Config.subscribePolling]] error occurs when getConfiguration in polling process,configurationName:{},version:{}", request.configuration(), request.clientConfigurationVersion(), e); + } + + LOGGER.debug("[[type=Capa.Config.subscribePolling]] subscribe polling task end,response:{}", resp); + + if (resp != null && !Objects.equals(resp.configurationVersion(), version)) { + fluxSink.next(resp); + } + // todo: make the polling frequency configurable + }, 0, 5, TimeUnit.SECONDS); + }) .publishOn(AwsCapaConfigurationScheduler.INSTANCE.configPublisherScheduler) .map(origin -> { GetConfigurationResponse resp = (GetConfigurationResponse) origin; diff --git a/capa-spi-aws-infrastructure/pom.xml b/capa-spi-aws-infrastructure/pom.xml index a9e1efb..63da965 100644 --- a/capa-spi-aws-infrastructure/pom.xml +++ b/capa-spi-aws-infrastructure/pom.xml @@ -23,7 +23,7 @@ capa-aws-parent group.rxcloud - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE capa-spi-aws-infrastructure @@ -37,11 +37,6 @@ capa-sdk-spi - - group.rxcloud - capa-foundation - - org.junit.jupiter diff --git a/capa-spi-aws-infrastructure/src/main/java/group/rxcloud/capa/spi/aws/infrastructure/AwsCapaEnvironment.java b/capa-spi-aws-infrastructure/src/main/java/group/rxcloud/capa/spi/aws/infrastructure/AwsCapaEnvironment.java index 1f9c3ee..6b3d627 100644 --- a/capa-spi-aws-infrastructure/src/main/java/group/rxcloud/capa/spi/aws/infrastructure/AwsCapaEnvironment.java +++ b/capa-spi-aws-infrastructure/src/main/java/group/rxcloud/capa/spi/aws/infrastructure/AwsCapaEnvironment.java @@ -16,8 +16,6 @@ */ package group.rxcloud.capa.spi.aws.infrastructure; -import group.rxcloud.capa.addons.foundation.CapaFoundation; -import group.rxcloud.capa.addons.foundation.FoundationType; import group.rxcloud.capa.infrastructure.CapaEnvironment; import group.rxcloud.capa.infrastructure.CapaProperties; @@ -41,8 +39,8 @@ public String getDeployRegion() { @Override public String getDeployEnv() { - // FIXME: 2021/12/15 use trip logic currently - return CapaFoundation.getEnv(FoundationType.TRIP); + String envKey = Settings.getEnvKey(); + return System.getProperty(envKey); } abstract static class Settings { diff --git a/capa-spi-aws-log/pom.xml b/capa-spi-aws-log/pom.xml index b29543d..33ca40a 100644 --- a/capa-spi-aws-log/pom.xml +++ b/capa-spi-aws-log/pom.xml @@ -23,11 +23,10 @@ capa-aws-parent group.rxcloud - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE capa-spi-aws-log - 1.11.13.4-alpha-1 2.16.0 @@ -40,16 +39,6 @@ group.rxcloud capa-spi-aws-infrastructure - 1.11.13.4-alpha-1 - - - - group.rxcloud - capa-id-generator - - - group.rxcloud - capa-cat diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLog4jAppender.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLog4jAppender.java index 0d486fd..4ff60f3 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLog4jAppender.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLog4jAppender.java @@ -16,24 +16,13 @@ */ package group.rxcloud.capa.spi.aws.log.appender; -import group.rxcloud.capa.infrastructure.hook.Mixer; -import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; import group.rxcloud.capa.spi.aws.log.filter.factory.LogOutputFactoryFilter; import group.rxcloud.capa.spi.aws.log.manager.CustomLogManager; import group.rxcloud.capa.spi.aws.log.manager.LogAppendManager; +import group.rxcloud.capa.spi.aws.log.service.LogMetrics; import group.rxcloud.capa.spi.log.CapaLog4jAppenderSpi; -import io.opentelemetry.api.common.AttributeKey; -import io.opentelemetry.api.common.Attributes; -import io.opentelemetry.api.metrics.LongCounter; -import io.opentelemetry.api.metrics.Meter; import org.apache.logging.log4j.core.LogEvent; import org.apache.logging.log4j.core.config.plugins.util.PluginManager; -import org.apache.logging.log4j.util.ReadOnlyStringMap; - -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; public class CapaAwsLog4jAppender extends CapaLog4jAppenderSpi { @@ -42,78 +31,28 @@ public class CapaAwsLog4jAppender extends CapaLog4jAppenderSpi { */ protected static final String LOG_LOG4J_APPENDER_ERROR_TYPE = "Log4jAppendLogsError"; - /** - * Number of counts each time. - */ - protected static final Integer COUNTER_NUM = 1; - - /** - * The namespace for logging error. - * TODO Set variables to common variables - */ - private static final String LOG_ERROR_NAMESPACE = "CloudWatchLogs"; - - /** - * The metric name for logging error. - * TODO Set variables to common variables - */ - private static final String LOG_ERROR_METRIC_NAME = "LogError"; - - private static final AtomicBoolean METRIC_INIT = new AtomicBoolean(false); - - /** - * Init an instance of {@link LongCounter}. - */ - protected static Optional LONG_COUNTER = Optional.empty(); - static { PluginManager.addPackage("group.rxcloud.capa.spi.aws.log.appender"); } - static Optional getCounterOpt() { - if (METRIC_INIT.get()) { - return LONG_COUNTER; - } - synchronized (METRIC_INIT) { - if (METRIC_INIT.compareAndSet(false, true)) { - Mixer.telemetryHooksNullable().ifPresent(telemetryHooks -> { - Meter meter = telemetryHooks.buildMeter(LOG_ERROR_NAMESPACE).block(); - LongCounter longCounter = meter.counterBuilder(LOG_ERROR_METRIC_NAME).build(); - LONG_COUNTER = Optional.ofNullable(longCounter); - }); - } - } - return LONG_COUNTER; - } - @Override public void appendLog(LogEvent event) { try { - if (event == null - || event.getLevel() == null - || event.getMessage() == null) { + if (event == null || event.getLevel() == null) { return; } - Optional capaLogLevel = CapaLogLevel.toCapaLogLevel(event.getLevel().name()); - if (capaLogLevel.isPresent() && LogOutputFactoryFilter.logCanOutput(capaLogLevel.get())) { - String message = event.getMessage().getFormattedMessage(); - ReadOnlyStringMap contextData = event.getContextData(); - Map MDCTags = contextData == null ? new HashMap<>() : contextData.toMap(); - LogAppendManager.appendLogs(message, MDCTags, event.getLoggerName(), event.getThreadName(), - event.getLevel().name(), event.getTimeMillis(), event.getThrown()); + + CapaLogEvent capaLogEvent = new CapaLogEvent(event); + if (LogOutputFactoryFilter.logCanOutput(capaLogEvent)) { + LogAppendManager.appendLogs(capaLogEvent); } } catch (Exception e) { try { CustomLogManager.error("CapaAwsLog4jAppender appender log error.", e); //Enhance function without affecting function - getCounterOpt().ifPresent(longCounter -> { - longCounter.bind(Attributes - .of(AttributeKey.stringKey(LOG_LOG4J_APPENDER_ERROR_TYPE), e.getClass().getName())) - .add(COUNTER_NUM); - }); + LogMetrics.recordLogError(LOG_LOG4J_APPENDER_ERROR_TYPE, e.getClass().getCanonicalName()); } catch (Throwable ex) { } - } } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLogbackAppender.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLogbackAppender.java index d161864..0f718ab 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLogbackAppender.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLogbackAppender.java @@ -17,21 +17,11 @@ package group.rxcloud.capa.spi.aws.log.appender; import ch.qos.logback.classic.spi.ILoggingEvent; -import ch.qos.logback.classic.spi.ThrowableProxy; -import group.rxcloud.capa.infrastructure.hook.Mixer; -import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; import group.rxcloud.capa.spi.aws.log.filter.factory.LogOutputFactoryFilter; import group.rxcloud.capa.spi.aws.log.manager.CustomLogManager; import group.rxcloud.capa.spi.aws.log.manager.LogAppendManager; +import group.rxcloud.capa.spi.aws.log.service.LogMetrics; import group.rxcloud.capa.spi.log.CapaLogbackAppenderSpi; -import io.opentelemetry.api.common.AttributeKey; -import io.opentelemetry.api.common.Attributes; -import io.opentelemetry.api.metrics.LongCounter; -import io.opentelemetry.api.metrics.Meter; - -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.atomic.AtomicBoolean; public class CapaAwsLogbackAppender extends CapaLogbackAppenderSpi { @@ -40,79 +30,24 @@ public class CapaAwsLogbackAppender extends CapaLogbackAppenderSpi { */ protected static final String LOG_LOGBACK_APPENDER_ERROR_TYPE = "LogbackAppendLogsError"; - /** - * Number of counts each time. - */ - protected static final Integer COUNTER_NUM = 1; - - /** - * The namespace for logging error. - * TODO Set variables to common variables - */ - private static final String LOG_ERROR_NAMESPACE = "CloudWatchLogs"; - - /** - * The metric name for logging error. - * TODO Set variables to common variables - */ - private static final String LOG_ERROR_METRIC_NAME = "LogError"; - - private static final AtomicBoolean METRIC_INIT = new AtomicBoolean(false); - - /** - * Init an instance of {@link LongCounter}. - */ - protected static Optional LONG_COUNTER = Optional.empty(); - - static Optional getCounterOpt() { - if (METRIC_INIT.get()) { - return LONG_COUNTER; - } - synchronized (METRIC_INIT) { - if (METRIC_INIT.compareAndSet(false, true)) { - Mixer.telemetryHooksNullable().ifPresent(telemetryHooks -> { - Meter meter = telemetryHooks.buildMeter(LOG_ERROR_NAMESPACE).block(); - LongCounter longCounter = meter.counterBuilder(LOG_ERROR_METRIC_NAME).build(); - LONG_COUNTER = Optional.ofNullable(longCounter); - }); - } - } - return LONG_COUNTER; - } - @Override public void appendLog(ILoggingEvent event) { try { if (event == null || event.getLevel() == null) { return; } - Optional capaLogLevel = CapaLogLevel.toCapaLogLevel(event.getLevel().levelStr); - if (capaLogLevel.isPresent() && LogOutputFactoryFilter.logCanOutput(capaLogLevel.get())) { - String message = event.getFormattedMessage(); - Map MDCTags = event.getMDCPropertyMap(); - LogAppendManager.appendLogs(message, MDCTags, event.getLoggerName(), event.getThreadName(), - event.getLevel().levelStr, event.getTimeStamp(), getThrowable(event)); + + CapaLogEvent capaLogEvent = new CapaLogEvent(event); + if (LogOutputFactoryFilter.logCanOutput(capaLogEvent)) { + LogAppendManager.appendLogs(capaLogEvent); } } catch (Exception e) { - CustomLogManager.error("CapaAwsLogbackAppender appender log error.", e); - getCounterOpt().ifPresent(longCounter -> { - try { - //Enhance function without affecting function - longCounter.bind(Attributes - .of(AttributeKey.stringKey(LOG_LOGBACK_APPENDER_ERROR_TYPE), e.getClass().getName())) - .add(COUNTER_NUM); - } catch (Throwable ex) { - } - }); - } - } - - private Throwable getThrowable(ILoggingEvent event) { - ThrowableProxy throwableProxy = (ThrowableProxy) event.getThrowableProxy(); - if (throwableProxy != null) { - return throwableProxy.getThrowable(); + try { + CustomLogManager.error("CapaAwsLogbackAppender appender log error.", e); + LogMetrics.recordLogError(LOG_LOGBACK_APPENDER_ERROR_TYPE, e.getClass().getCanonicalName()); + } catch (Throwable ex) { + } } - return null; } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaLogEvent.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaLogEvent.java new file mode 100644 index 0000000..9cd8d98 --- /dev/null +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaLogEvent.java @@ -0,0 +1,179 @@ +/* + * 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 group.rxcloud.capa.spi.aws.log.appender; + +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.classic.spi.ThrowableProxy; +import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.util.ReadOnlyStringMap; + +import javax.annotation.Nonnull; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +/** + * Capa log event. + */ +public class CapaLogEvent { + + /** + * Tag identifier prefix. + */ + private static final String TAG_PREFIX = "[["; + + /** + * Tag identifier suffix. + */ + private static final String TAG_SUFFIX = "]]"; + + private Optional capaLogLevel; + + private String message; + + private String loggerName; + + private String threadName; + + private long time; + + private Throwable throwable; + + private Map tags; + + private boolean throttle; + + public CapaLogEvent(LogEvent event) { + capaLogLevel = CapaLogLevel.toCapaLogLevel(event.getLevel().name()); + String originMessage = event.getMessage().getFormattedMessage(); + message = deleteTags(originMessage); + loggerName = event.getLoggerName(); + throwable = event.getThrown(); + threadName = event.getThreadName(); + time = event.getTimeMillis(); + ReadOnlyStringMap contextData = event.getContextData(); + Map MDCTags = contextData == null ? null : contextData.toMap(); + tags = mergeTags(originMessage, MDCTags); + } + + public CapaLogEvent(String loggerName, CapaLogLevel level, String message, Throwable throwable) { + capaLogLevel = Optional.ofNullable(level); + this.message = deleteTags(message); + this.loggerName = loggerName; + this.throwable = throwable; + threadName = Thread.currentThread().getName(); + time = System.currentTimeMillis(); + tags = mergeTags(message, null); + } + + public CapaLogEvent(ILoggingEvent event) { + capaLogLevel = CapaLogLevel.toCapaLogLevel(event.getLevel().levelStr); + loggerName = event.getLoggerName(); + String originMessage = event.getFormattedMessage(); + message = deleteTags(originMessage); + ThrowableProxy throwableProxy = (ThrowableProxy) event.getThrowableProxy(); + if (throwableProxy != null) { + throwable = throwableProxy.getThrowable(); + } + threadName = event.getThreadName(); + time = event.getTimeStamp(); + tags = mergeTags(originMessage, event.getMDCPropertyMap()); + } + + private static String deleteTags(String message) { + if (message.startsWith(TAG_PREFIX)) { + int tagsEndIndex = message.indexOf(TAG_SUFFIX); + if (tagsEndIndex > 0) { + return message.substring(tagsEndIndex + 2); + } + } + return message; + } + + private static Map mergeTags(String message, Map MDCTags) { + Map tags = new HashMap<>(); + if (message.startsWith(TAG_PREFIX)) { + int tagsEndIndex = message.indexOf(TAG_SUFFIX); + if (tagsEndIndex > 0) { + parseTags(message, tagsEndIndex, tags); + } + } + if (MDCTags != null && !MDCTags.isEmpty()) { + tags.putAll(MDCTags); + } + return tags; + } + + private static void parseTags(String message, int tagsEndIndex, Map tags) { + int tagStart = 2; + while (tagStart < tagsEndIndex) { + int tagEnd = message.indexOf(',', tagStart); + if (tagEnd < 0 || tagEnd > tagsEndIndex) { + tagEnd = tagsEndIndex; + } + int equalIndex = message.indexOf('=', tagStart); + if (equalIndex > tagStart && equalIndex < tagEnd - 1) { + String key = message.substring(tagStart, equalIndex); + String value = message.substring(equalIndex + 1, tagEnd); + tags.put(key, value); + } + tagStart = tagEnd + 1; + } + } + + public boolean isThrottle() { + return throttle; + } + + public void setThrottle(boolean throttle) { + this.throttle = throttle; + } + + public Map getTags() { + return tags; + } + + public String getThreadName() { + return threadName; + } + + public long getTime() { + return time; + } + + public Optional getCapaLogLevel() { + return capaLogLevel; + } + + @Nonnull + public String getLogLevel() { + return capaLogLevel.map(CapaLogLevel::getLevelName).orElse("UNDEFINED"); + } + + public String getMessage() { + return message; + } + + public String getLoggerName() { + return loggerName; + } + + public Throwable getThrowable() { + return throwable; + } +} diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/CapaComponentLogConfiguration.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/CapaComponentLogConfiguration.java index 5f2ce9b..07f41cd 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/CapaComponentLogConfiguration.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/CapaComponentLogConfiguration.java @@ -16,7 +16,6 @@ */ package group.rxcloud.capa.spi.aws.log.configuration; -import com.google.common.collect.Lists; import group.rxcloud.capa.component.CapaLogProperties; import group.rxcloud.capa.infrastructure.hook.ConfigurationHooks; import group.rxcloud.capa.infrastructure.hook.Mixer; @@ -31,73 +30,73 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReferenceArray; -import java.util.function.Consumer; public class CapaComponentLogConfiguration { + private static final String CAPA_COMPONENT_LOG_CONFIGURATION_FILE_NAME = "capa-component-log-configuration.properties"; - private static CapaComponentLogConfiguration instance; - private final Logger log = LoggerFactory.getLogger(CapaComponentLogConfiguration.class); + + private static final AtomicBoolean INIT = new AtomicBoolean(false); + private static final Object lock = new Object(); + + private static Optional instance = Optional.empty(); + + private final Logger log = LoggerFactory.getLogger(CapaComponentLogConfiguration.class); + private final AtomicReferenceArray capaComponentLogConfigurationProperties; - private final List>> customConfigCallbackList; + + private final List callbackList; + private volatile Map capaComponentLogConfiguration; private CapaComponentLogConfiguration() { capaComponentLogConfigurationProperties = new AtomicReferenceArray<>(2); capaComponentLogConfiguration = new HashMap<>(); - customConfigCallbackList = new ArrayList<>(); + callbackList = new ArrayList<>(); + callbackList.add(new EffectiveTimeChecker()); Mixer.configurationHooksNullable().ifPresent(hooks -> { // subscribe capa-compoment-log-configuration.properties - List appIds = Lists.newArrayList(hooks.defaultConfigurationAppId(), CapaLogProperties.Settings.getCenterConfigAppId()); + List appIds = new ArrayList<>(2); + appIds.add(hooks.defaultConfigurationAppId()); + appIds.add(CapaLogProperties.Settings.getCenterConfigAppId()); for (int i = 0; i < appIds.size(); i++) { try { subscribeCapaComponentLogConfigurationByAppId(hooks, appIds.get(i), i); } catch (Throwable throwable) { log.warn("Fail to subscribe config for app id " + appIds.get(i) + ", index " + i, throwable); + capaComponentLogConfigurationProperties.set(i, null); } } }); } - public static CapaComponentLogConfiguration getInstance() { - if (instance != null) { + + public static Optional getInstanceOpt() { + if (INIT.get()) { return instance; } + synchronized (lock) { - if (instance == null) { - instance = new CapaComponentLogConfiguration(); + if (INIT.compareAndSet(false, true)) { + instance = Optional.of(new CapaComponentLogConfiguration()); } } return instance; } - public void registerConfigCallback(Consumer> consumer) { - synchronized (customConfigCallbackList) { - customConfigCallbackList.add(consumer); - } - } - public boolean containsKey(String key) { - try { - return capaComponentLogConfiguration != null - && capaComponentLogConfiguration.containsKey(key); - } catch (Exception e) { - return false; - } + return capaComponentLogConfiguration.containsKey(key); } public String get(String key) { - try { - return capaComponentLogConfiguration == null - ? "" - : capaComponentLogConfiguration.get(key); - } catch (Exception e) { - return ""; - } + return capaComponentLogConfiguration.get(key); } - private void subscribeCapaComponentLogConfigurationByAppId(ConfigurationHooks configurationHooks, String appId, int index) { + private void subscribeCapaComponentLogConfigurationByAppId(ConfigurationHooks configurationHooks, String appId, + int index) { String storeName = configurationHooks.registryStoreNames().get(0); Flux> configFlux = configurationHooks.subscribeConfiguration( @@ -109,11 +108,7 @@ private void subscribeCapaComponentLogConfigurationByAppId(ConfigurationHooks co "", TypeRef.get(Map.class)); - // FIXME: 2021/12/3 random callback? configFlux.subscribe(resp -> { - for (Consumer> subConfigurationRespConsumer : customConfigCallbackList) { - subConfigurationRespConsumer.accept(resp); - } synchronized (lock) { if (!resp.getItems().isEmpty()) { capaComponentLogConfigurationProperties.set(index, resp.getItems().get(0).getContent()); @@ -121,15 +116,26 @@ private void subscribeCapaComponentLogConfigurationByAppId(ConfigurationHooks co capaComponentLogConfigurationProperties.set(index, null); } - Map merged = new HashMap<>(); + Map newConfig = new HashMap<>(); for (int i = 0; i < capaComponentLogConfigurationProperties.length(); i++) { Map item = capaComponentLogConfigurationProperties.get(i); if (item != null) { - item.forEach((k, v) -> merged.putIfAbsent(String.valueOf(k), String.valueOf(v))); + item.forEach((k, v) -> newConfig.putIfAbsent(String.valueOf(k), String.valueOf(v))); } } - this.capaComponentLogConfiguration = merged; + + Map oldConfig = capaComponentLogConfiguration; + capaComponentLogConfiguration = newConfig; + + for (ConfigChangedCallback callback : callbackList) { + callback.onChange(oldConfig, newConfig); + } } }); } + + @FunctionalInterface + interface ConfigChangedCallback { + void onChange(Map oldConfig, Map newConfig); + } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/EffectiveTimeChecker.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/EffectiveTimeChecker.java new file mode 100644 index 0000000..1e92b8d --- /dev/null +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/EffectiveTimeChecker.java @@ -0,0 +1,48 @@ +/* + * 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 group.rxcloud.capa.spi.aws.log.configuration; + +import java.util.Map; + +/** + * Checker of output level effective time. + */ +public class EffectiveTimeChecker implements CapaComponentLogConfiguration.ConfigChangedCallback { + + private static volatile long levelChangedStart = -1; + + private static boolean isChanged(Map oldConfig, Map newConfig) { + return !oldConfig.getOrDefault(LogConfig.LevelConfig.OUTPUT_LEVEL.configKey, "") + .equals(newConfig.getOrDefault(LogConfig.LevelConfig.OUTPUT_LEVEL.configKey, "")); + } + + public static boolean isOutputLevelEffective() { + if (levelChangedStart > 0) { + long end = levelChangedStart + LogConfig.TimeConfig.OUTPUT_LOG_EFFECTIVE_TIME.get(); + return end > System.currentTimeMillis(); + } + return true; + } + + @Override + public void onChange(Map oldConfig, Map newConfig) { + boolean isLevelChanged = isChanged(oldConfig, newConfig); + if (isLevelChanged) { + levelChangedStart = System.currentTimeMillis(); + } + } +} diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/LogConfig.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/LogConfig.java new file mode 100644 index 0000000..16964c5 --- /dev/null +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/configuration/LogConfig.java @@ -0,0 +1,132 @@ +/* + * 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 group.rxcloud.capa.spi.aws.log.configuration; + +import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +/** + * Log common configs. + */ +public interface LogConfig { + + T get(); + + enum BoolConfig implements LogConfig { + LOG_SWITCH("logSwitch", true); + + String configKey; + + boolean defaultValue; + + BoolConfig(String configKey, boolean defaultValue) { + this.configKey = configKey; + this.defaultValue = defaultValue; + } + + @Override + public Boolean get() { + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + if (configuration.isPresent()) { + String value = configuration.get().get(configKey); + return value == null ? defaultValue : Boolean.valueOf(value); + } + return defaultValue; + } + } + + enum IntConfig implements LogConfig { + ALERT_LOG_COUNT("alertLogCount", 100); + + String configKey; + + int defaultValue; + + IntConfig(String configKey, int defaultValue) { + this.configKey = configKey; + this.defaultValue = defaultValue; + } + + @Override + public Integer get() { + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + if (configuration.isPresent()) { + String value = configuration.get().get(configKey); + return value == null ? defaultValue : Integer.parseInt(value); + } + return defaultValue; + } + } + + enum TimeConfig implements LogConfig { + + OUTPUT_LOG_EFFECTIVE_TIME("outputLogEffectiveTime", 30), + ALERT_LOG_COUNT_TIME("alertLogCountMinutes", 5), + ALERT_LOG_IGNORE_IGNORE_TIME("alertLogIgnoreMinutes", 60); + + String configKey; + + int defaultValue; + + TimeConfig(String configKey, int defaultValue) { + this.configKey = configKey; + this.defaultValue = defaultValue; + } + + @Override + public Long get() { + int target = defaultValue; + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + if (configuration.isPresent()) { + String value = configuration.get().get(configKey); + if (value != null) { + target = Integer.parseInt(value); + } + } + return TimeUnit.MINUTES.toMillis(target); + } + } + + enum LevelConfig implements LogConfig { + + OUTPUT_LEVEL("outputLogLevel", CapaLogLevel.ERROR), + DEFAULT_OUT_PUT_LEVEL("defaultOutputLogLevel", CapaLogLevel.ERROR), + ALERT_LOG_LEVEL("alertLogLevel", CapaLogLevel.ERROR); + + String configKey; + + CapaLogLevel defaultValue; + + LevelConfig(String configKey, CapaLogLevel defaultValue) { + this.configKey = configKey; + this.defaultValue = defaultValue; + } + + public Optional getOpt() { + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + return configuration + .flatMap(logConfiguration -> CapaLogLevel.toCapaLogLevel(logConfiguration.get(configKey))); + } + + @Override + public CapaLogLevel get() { + return getOpt().orElse(defaultValue); + } + } +} diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/enums/CapaLogLevel.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/enums/CapaLogLevel.java index ee92117..7fdc0ad 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/enums/CapaLogLevel.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/enums/CapaLogLevel.java @@ -46,20 +46,24 @@ public enum CapaLogLevel { /** * Convert logLevelArg to {@link CapaLogLevel} + * @return CapaLogLevel */ public static Optional toCapaLogLevel(String logLevelArg) { - return Arrays.stream(CapaLogLevel.values()) + return Arrays.stream(values()) .filter(logLevel -> logLevel.levelName.equalsIgnoreCase(logLevelArg)) .findAny(); } /** - * Get level. + * @return log level. */ public int getLevel() { return level; } + /** + * @return log level name. + */ public String getLevelName() { return levelName; } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/LogOutputFilter.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/LogOutputFilter.java index 5728220..0590e2b 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/LogOutputFilter.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/LogOutputFilter.java @@ -16,9 +16,9 @@ */ package group.rxcloud.capa.spi.aws.log.filter; -import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; public interface LogOutputFilter { - boolean logCanOutput(CapaLogLevel level); + boolean logCanOutput(CapaLogEvent event); } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/factory/LogOutputFactoryFilter.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/factory/LogOutputFactoryFilter.java index 0182da3..5aaee8b 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/factory/LogOutputFactoryFilter.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/factory/LogOutputFactoryFilter.java @@ -16,23 +16,29 @@ */ package group.rxcloud.capa.spi.aws.log.filter.factory; -import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; import group.rxcloud.capa.spi.aws.log.filter.LogOutputFilter; +import group.rxcloud.capa.spi.aws.log.filter.logoutput.LogOutputCountFilter; import group.rxcloud.capa.spi.aws.log.filter.logoutput.LogOutputLevelFilter; import group.rxcloud.capa.spi.aws.log.filter.logoutput.LogOutputSwitchFilter; -import group.rxcloud.capa.spi.aws.log.filter.logoutput.LogOutputTimeFilter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.Collections; import java.util.List; -import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; -public class LogOutputFactoryFilter { +public final class LogOutputFactoryFilter { + private static final AtomicBoolean FILTER_INIT = new AtomicBoolean(false); + private static final Logger log = LoggerFactory.getLogger(LogOutputFactoryFilter.class); - private static Optional> logOutputFilterList = Optional.empty(); + + private static volatile List logOutputFilterList = Collections.emptyList(); + + private LogOutputFactoryFilter() { + } /** * To judge whether the output can be output, several conditions need to be considered. @@ -58,21 +64,19 @@ public class LogOutputFactoryFilter { * first set the error level to error or warn, * and then change it back to info to take effect * - * @param outputLogLevel - * @return + * @param event event + * @return whether log can output */ - public static boolean logCanOutput(CapaLogLevel outputLogLevel) { - if (getLogOutputFilterList().isPresent()) { - for (LogOutputFilter logOutputFilter : logOutputFilterList.get()) { - if (!logOutputFilter.logCanOutput(outputLogLevel)) { - return false; - } + public static boolean logCanOutput(CapaLogEvent event) { + for (LogOutputFilter logOutputFilter : getLogOutputFilterList()) { + if (!logOutputFilter.logCanOutput(event)) { + return false; } } return true; } - private static Optional> getLogOutputFilterList() { + private static List getLogOutputFilterList() { if (FILTER_INIT.get()) { return logOutputFilterList; } @@ -82,8 +86,8 @@ private static Optional> getLogOutputFilterList() { List logOutputFilters = new ArrayList<>(); logOutputFilters.add(new LogOutputSwitchFilter()); logOutputFilters.add(new LogOutputLevelFilter()); - logOutputFilters.add(new LogOutputTimeFilter()); - logOutputFilterList = Optional.of(logOutputFilters); + logOutputFilters.add(new LogOutputCountFilter()); + logOutputFilterList = logOutputFilters; } catch (Throwable e) { log.error("Create logOutputFilter error.", e); } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputCountFilter.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputCountFilter.java new file mode 100644 index 0000000..76d6f0c --- /dev/null +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputCountFilter.java @@ -0,0 +1,170 @@ +/* + * 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 group.rxcloud.capa.spi.aws.log.filter.logoutput; + +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; +import group.rxcloud.capa.spi.aws.log.configuration.LogConfig; +import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; +import group.rxcloud.capa.spi.aws.log.filter.LogOutputFilter; +import group.rxcloud.capa.spi.aws.log.manager.CustomLogManager; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.util.Map; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +public class LogOutputCountFilter implements LogOutputFilter { + + @Override + public boolean logCanOutput(CapaLogEvent event) { + return LogLimiter.logsCountLimit(event); + } + + static final class LogLimiter { + + private static final ConcurrentHashMap COUNTER_MAP = new ConcurrentHashMap<>(); + + private static final LinkedBlockingQueue KEYS = new LinkedBlockingQueue<>(); + + private static final Timer CLEARNER = new Timer("log-alert-cleaner", true); + + static { + CLEARNER.scheduleAtFixedRate(new TimerTask() { + @Override + public void run() { + try { + while (!KEYS.isEmpty() && isExpired(COUNTER_MAP.get(KEYS.peek()))) { + COUNTER_MAP.remove(KEYS.poll()); + } + } catch (Throwable e) { + CustomLogManager.warn("Fail to clean alert log counter.", e); + } + } + }, TimeUnit.MINUTES.toMillis(1L), TimeUnit.MINUTES.toMillis(1L)); + } + + private LogLimiter() { + } + + private static boolean isExpired(OutputCount outputCount) { + long countMillis = LogConfig.TimeConfig.ALERT_LOG_COUNT_TIME.get(); + long alertMillis = LogConfig.TimeConfig.ALERT_LOG_IGNORE_IGNORE_TIME.get(); + + return outputCount.alertTime() > alertMillis || outputCount.countTime() > countMillis; + } + + public static boolean logsCountLimit(CapaLogEvent event) { + CapaLogLevel level = event.getCapaLogLevel().orElse(null); + String loggerName = event.getLoggerName(); + String message = event.getMessage(); + Throwable ex = event.getThrowable(); + Map tags = event.getTags(); + CapaLogLevel restrictLevel = LogConfig.LevelConfig.ALERT_LOG_LEVEL.get(); + if (level.getLevel() < restrictLevel.getLevel()) { + return true; + } + + long key = encode(loggerName, message, tags, ex); + OutputCount outputCount = COUNTER_MAP.computeIfAbsent(key, k -> { + KEYS.offer(k); + return new OutputCount(); + }); + + // check if the count is expired. + if (isExpired(outputCount)) { + outputCount.clear(); + } + + int restrictCount = LogConfig.IntConfig.ALERT_LOG_COUNT.get(); + int count = outputCount.increamentAndGet(); + if (count < restrictCount) { + return true; + } + + // alert + if (outputCount.startAlert()) { + event.setThrottle(true); + return true; + } + + return false; + } + + + private static long encode(String loggerName, String message, Map tags, Throwable ex) { + StringWriter stringWriter = new StringWriter(256); + PrintWriter writer = new PrintWriter(stringWriter); + writer.print(loggerName); + writer.print(message); + if (tags != null && !tags.isEmpty()) { + tags.forEach((k, v) -> { + writer.print(k); + writer.print(v); + }); + } + if (ex != null) { + writer.print(ex.getMessage()); + ex.printStackTrace(writer); + } + return stringWriter.toString().hashCode(); + } + } + + static class OutputCount { + + final AtomicInteger count = new AtomicInteger(); + + long startTimeMillis = System.currentTimeMillis(); + + volatile long alertStartTimeMillis; + + void clear() { + count.set(0); + alertStartTimeMillis = 0L; + startTimeMillis = System.currentTimeMillis(); + } + + long countTime() { + return System.currentTimeMillis() - startTimeMillis; + } + + long alertTime() { + return alertStartTimeMillis > 0L ? System.currentTimeMillis() - alertStartTimeMillis : 0L; + } + + boolean startAlert() { + if (alertStartTimeMillis == 0L) { + synchronized (count) { + if (alertStartTimeMillis == 0L) { + alertStartTimeMillis = System.currentTimeMillis(); + return true; + } + } + } + return false; + } + + int increamentAndGet() { + return count.incrementAndGet(); + } + } +} diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputLevelFilter.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputLevelFilter.java index 7ba91ea..70f9b6f 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputLevelFilter.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputLevelFilter.java @@ -16,33 +16,32 @@ */ package group.rxcloud.capa.spi.aws.log.filter.logoutput; -import group.rxcloud.capa.spi.aws.log.configuration.CapaComponentLogConfiguration; +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; +import group.rxcloud.capa.spi.aws.log.configuration.EffectiveTimeChecker; +import group.rxcloud.capa.spi.aws.log.configuration.LogConfig; import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; import group.rxcloud.capa.spi.aws.log.filter.LogOutputFilter; import java.util.Optional; public class LogOutputLevelFilter implements LogOutputFilter { - private static final String OUTPUT_LOG_LEVEL_NAME = "outputLogLevel"; - private final Optional capaComponentLogConfiguration = Optional.ofNullable(CapaComponentLogConfiguration.getInstance()); - - public LogOutputLevelFilter() { - } @Override - public boolean logCanOutput(CapaLogLevel outputLogLevel) { - // 1. Check whether the output log level is higher than or equal to the number of log output levels configured by the application. - // If it is lower than the configuration, return false directly. - // Whether the log level is higher than or equal to the log output level configured by the application. - if (capaComponentLogConfiguration.isPresent() - && capaComponentLogConfiguration.get().containsKey(OUTPUT_LOG_LEVEL_NAME)) { - Optional capaLogLevel = CapaLogLevel.toCapaLogLevel(capaComponentLogConfiguration.get().get(OUTPUT_LOG_LEVEL_NAME)); - if (capaLogLevel.isPresent() - && outputLogLevel.getLevel() < capaLogLevel.get().getLevel()) { - return false; - } + public boolean logCanOutput(CapaLogEvent event) { + // Check whether the output log level is higher than or equal to the log output level configured by capa-log, and return true if it is higher than or equal to. + Optional currentLevelOpt = event.getCapaLogLevel(); + if (!currentLevelOpt.isPresent()) { + return true; + } + + CapaLogLevel currentLevel = currentLevelOpt.get(); + Optional outputLevel = LogConfig.LevelConfig.OUTPUT_LEVEL.getOpt(); + // if the output level is effective, skip the default level check. + if (outputLevel.isPresent() && EffectiveTimeChecker.isOutputLevelEffective()) { + return currentLevel.getLevel() >= outputLevel.get().getLevel(); } - return true; + CapaLogLevel defaultOutputLevel = LogConfig.LevelConfig.DEFAULT_OUT_PUT_LEVEL.get(); + return currentLevel.getLevel() >= defaultOutputLevel.getLevel(); } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputSwitchFilter.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputSwitchFilter.java index 0d48f48..2a350ed 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputSwitchFilter.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputSwitchFilter.java @@ -16,25 +16,19 @@ */ package group.rxcloud.capa.spi.aws.log.filter.logoutput; -import group.rxcloud.capa.spi.aws.log.configuration.CapaComponentLogConfiguration; -import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; +import group.rxcloud.capa.spi.aws.log.configuration.LogConfig; import group.rxcloud.capa.spi.aws.log.filter.LogOutputFilter; -import java.util.Optional; - public class LogOutputSwitchFilter implements LogOutputFilter { + /** * Dynamically adjust the log level switch name. */ - private static final String LOG_SWITCH_NAME = "logSwitch"; - - private final Optional capaComponentLogConfiguration = Optional.ofNullable(CapaComponentLogConfiguration.getInstance()); @Override - public boolean logCanOutput(CapaLogLevel level) { + public boolean logCanOutput(CapaLogEvent event) { // Determine whether the log output switch is turned on, if it is turned off, the log will not be output. - return !capaComponentLogConfiguration.isPresent() - || !capaComponentLogConfiguration.get().containsKey(LOG_SWITCH_NAME) - || !String.valueOf(Boolean.FALSE).equalsIgnoreCase(capaComponentLogConfiguration.get().get(LOG_SWITCH_NAME)); + return LogConfig.BoolConfig.LOG_SWITCH.get(); } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputTimeFilter.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputTimeFilter.java deleted file mode 100644 index 4f4c305..0000000 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/filter/logoutput/LogOutputTimeFilter.java +++ /dev/null @@ -1,105 +0,0 @@ -/* - * 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 group.rxcloud.capa.spi.aws.log.filter.logoutput; - -import group.rxcloud.capa.spi.aws.log.configuration.CapaComponentLogConfiguration; -import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; -import group.rxcloud.capa.spi.aws.log.filter.LogOutputFilter; -import group.rxcloud.cloudruntimes.domain.core.configuration.SubConfigurationResp; - -import java.util.Calendar; -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; -import java.util.function.Consumer; - -public class LogOutputTimeFilter implements LogOutputFilter { - private static final String OUTPUT_LOG_LEVEL_CONFIG_KEY = "outputLogLevel"; - private static final Integer DEFAULT_OUTPUT_LOG_EFFECTIVE_TIME = 30; - private static final String OUTPUT_LOG_EFFECTIVE_TIME_CONFIG_KEY = "outputLogEffectiveTime"; - private static final CapaLogLevel DEFAULT_OUTPUT_LOG_LEVEL = CapaLogLevel.ERROR; - private static final String DEFAULT_OUTPUT_LOG_LEVEL_NAME = "defaultOutputLogLevel"; - private final Optional capaComponentLogConfiguration = Optional.ofNullable(CapaComponentLogConfiguration.getInstance()); - private volatile Long currentOutputLogValidTime; - private volatile CapaLogLevel currentLogLevel = CapaLogLevel.ERROR; - private volatile Integer currentOutputLogLevelEffectiveMin; - - - public LogOutputTimeFilter() { - Consumer> capaComponentLogConfigurationConsumer = resp -> { - Map config = new HashMap<>(); - if (!resp.getItems().isEmpty()) { - config = resp.getItems().get(0).getContent(); - } - // update current output log level valid time - if (config.isEmpty()) { - this.currentLogLevel = CapaLogLevel.ALL; - } - if (!config.isEmpty() && config.containsKey(OUTPUT_LOG_LEVEL_CONFIG_KEY)) { - CapaLogLevel.toCapaLogLevel(config.get(OUTPUT_LOG_LEVEL_CONFIG_KEY)) - .ifPresent(outputLogLevel -> { - if (!this.currentLogLevel.equals(outputLogLevel)) { - this.currentLogLevel = outputLogLevel; - this.updateOutputLogValidTime(); - } - }); - } - // update output log level and the valid time - if (!config.isEmpty() && config.containsKey(OUTPUT_LOG_EFFECTIVE_TIME_CONFIG_KEY)) { - Optional.ofNullable(Integer.parseInt(config.get(OUTPUT_LOG_EFFECTIVE_TIME_CONFIG_KEY))) - .ifPresent(outputLogEffectiveTime -> { - if (!this.currentOutputLogLevelEffectiveMin.equals(outputLogEffectiveTime)) { - this.currentOutputLogLevelEffectiveMin = outputLogEffectiveTime; - this.updateOutputLogValidTime(); - } - }); - } - }; - capaComponentLogConfiguration.ifPresent(capaComponentLogConfiguration -> { - capaComponentLogConfiguration.registerConfigCallback(capaComponentLogConfigurationConsumer); - }); - this.updateOutputLogValidTime(); - } - - private void updateOutputLogValidTime() { - this.currentOutputLogLevelEffectiveMin = DEFAULT_OUTPUT_LOG_EFFECTIVE_TIME; - if (CapaComponentLogConfiguration.getInstance() != null - && CapaComponentLogConfiguration.getInstance().containsKey(OUTPUT_LOG_EFFECTIVE_TIME_CONFIG_KEY)) { - this.currentOutputLogLevelEffectiveMin = Integer.parseInt(CapaComponentLogConfiguration.getInstance().get(OUTPUT_LOG_EFFECTIVE_TIME_CONFIG_KEY)); - } - - Calendar instance = Calendar.getInstance(); - instance.add(Calendar.MINUTE, this.currentOutputLogLevelEffectiveMin); - currentOutputLogValidTime = instance.getTimeInMillis(); - } - - @Override - public boolean logCanOutput(CapaLogLevel outputLogLevel) { - // Check whether the output log level is higher than or equal to the log output level configured by capa-log, and return true if it is higher than or equal to. - if (capaComponentLogConfiguration.isPresent() - && capaComponentLogConfiguration.get().containsKey(DEFAULT_OUTPUT_LOG_LEVEL_NAME)) { - Optional capaLogLevel = CapaLogLevel.toCapaLogLevel(capaComponentLogConfiguration.get().get(DEFAULT_OUTPUT_LOG_LEVEL_NAME)); - if (!capaLogLevel.isPresent()) { - capaLogLevel = Optional.of(DEFAULT_OUTPUT_LOG_LEVEL); - } - if (outputLogLevel.getLevel() >= capaLogLevel.get().getLevel()) { - return true; - } - } - return System.currentTimeMillis() < currentOutputLogValidTime; - } -} diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/handle/MessageSender.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/handle/MessageSender.java index 9800766..7105c05 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/handle/MessageSender.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/handle/MessageSender.java @@ -38,32 +38,46 @@ import java.util.Optional; import java.util.Random; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; public class MessageSender extends Thread { + private static final int MAX_COUNT_PER_CHUNK = 100; + private static final long WAIT_INTERVAL = 20L; + private static final int MAX_SIZE_PER_CHUNK = 1024 * 1024; + private static final String PUT_LOG_EVENTS_RESOURCE_NAME = "CloudWatchLogs.putLogEvents"; + private static final String MESSAGE_SENDER_ERROR_NAMESPACE = "LogMessageSenderError"; + private static final String MESSAGE_SENDER_ERROR_METRIC_NAME = "LogsSenderError"; + private static final String LOG_STREAM_COUNT_NAME = "logStreamCount"; + private static final int DEFAULT_MAX_RULE_COUNT = 10; + private static final String CLOUD_WATCH_AGENT_SWITCH_NAME = "cloudWatchAgentSwitch"; + private static Optional LONG_COUNTER = Optional.empty(); static { initFlowRules(); Mixer.telemetryHooksNullable() - .ifPresent(telemetryHooks -> { - Meter meter = telemetryHooks.buildMeter(MESSAGE_SENDER_ERROR_NAMESPACE).block(); - LongCounter longCounter = meter.counterBuilder(MESSAGE_SENDER_ERROR_METRIC_NAME).build(); - LONG_COUNTER = Optional.ofNullable(longCounter); - }); + .ifPresent(telemetryHooks -> { + Meter meter = telemetryHooks.buildMeter(MESSAGE_SENDER_ERROR_NAMESPACE).block(); + LongCounter longCounter = meter.counterBuilder(MESSAGE_SENDER_ERROR_METRIC_NAME).build(); + LONG_COUNTER = Optional.ofNullable(longCounter); + }); } private final ChunkQueue chunkQueue; + private final LinkedList readCompressedChunk; + private volatile boolean running = true; + private volatile CountDownLatch shutdownLatch; public MessageSender(ChunkQueue chunkQueue) { @@ -73,11 +87,14 @@ public MessageSender(ChunkQueue chunkQueue) { private static void initFlowRules() { List flowRules = new ArrayList<>(); - int ruleCount = CapaComponentLogConfiguration.getInstance().containsKey(LOG_STREAM_COUNT_NAME) - ? Integer.parseInt(LOG_STREAM_COUNT_NAME) - : DEFAULT_MAX_RULE_COUNT; + AtomicInteger ruleCount = new AtomicInteger(DEFAULT_MAX_RULE_COUNT); + CapaComponentLogConfiguration.getInstanceOpt().ifPresent(configuration -> { + if (configuration.containsKey(LOG_STREAM_COUNT_NAME)) { + ruleCount.set(Integer.parseInt(configuration.get(LOG_STREAM_COUNT_NAME))); + } + }); - for (int i = 0; i < ruleCount; i++) { + for (int i = 0; i < ruleCount.get(); i++) { FlowRule flowRule = new FlowRule(); flowRule.setResource(PUT_LOG_EVENTS_RESOURCE_NAME + "_" + i); flowRule.setGrade(RuleConstant.FLOW_GRADE_QPS); @@ -105,8 +122,9 @@ public void run() { } catch (Throwable throwable) { CustomLogManager.error("MessageSender build chunk error.", throwable); LONG_COUNTER.ifPresent(longCounter -> { - longCounter.bind(Attributes.of(AttributeKey.stringKey("BuildCompressedChunkError"), throwable.getClass().getName())) - .add(1); + longCounter.bind(Attributes + .of(AttributeKey.stringKey("BuildCompressedChunkError"), throwable.getClass().getName())) + .add(1); }); } @@ -126,8 +144,10 @@ public void run() { } private void putLogToCloudWatch(List logMessages) { - if (!CapaComponentLogConfiguration.getInstance().containsKey(CLOUD_WATCH_AGENT_SWITCH_NAME) - || Boolean.TRUE.toString().equalsIgnoreCase(CapaComponentLogConfiguration.getInstance().get(CLOUD_WATCH_AGENT_SWITCH_NAME))) { + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + if (!configuration.isPresent() + || !configuration.get().containsKey(CLOUD_WATCH_AGENT_SWITCH_NAME) + || Boolean.TRUE.toString().equalsIgnoreCase(configuration.get().get(CLOUD_WATCH_AGENT_SWITCH_NAME))) { // put logs by agent putLogsByAgent(logMessages); } else { @@ -154,8 +174,9 @@ private void putLogsByApi(List logMessages) { } catch (Throwable throwable) { CustomLogManager.error("MessageSender send message error.", throwable); LONG_COUNTER.ifPresent(longCounter -> { - longCounter.bind(Attributes.of(AttributeKey.stringKey("SenderPutLogEventsError"), throwable.getClass().getName())) - .add(1); + longCounter.bind(Attributes + .of(AttributeKey.stringKey("SenderPutLogEventsError"), throwable.getClass().getName())) + .add(1); }); } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/CustomLogManager.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/CustomLogManager.java index 55f420a..7c2d0a7 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/CustomLogManager.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/CustomLogManager.java @@ -16,64 +16,22 @@ */ package group.rxcloud.capa.spi.aws.log.manager; -import com.google.gson.Gson; +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel; -import java.util.Map; - // TODO upgrade public class CustomLogManager { - private static final Gson GSON = new Gson(); - - public static void trace(String message) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.TRACE.name(), null); - System.out.println(GSON.toJson(logMessages)); - } - - public static void trace(String message, Throwable throwable) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.TRACE.name(), throwable); - System.out.println(GSON.toJson(logMessages)); - } - - public static void debug(String message) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.DEBUG.name(), null); - System.out.println(GSON.toJson(logMessages)); - } - - public static void debug(String message, Throwable throwable) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.DEBUG.name(), throwable); - System.out.println(GSON.toJson(logMessages)); - } - - public static void info(String message) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.INFO.name(), null); - System.out.println(GSON.toJson(logMessages)); - } - - public static void info(String message, Throwable throwable) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.INFO.name(), throwable); - System.out.println(GSON.toJson(logMessages)); - } - - public static void warn(String message) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.WARN.name(), null); - System.out.println(GSON.toJson(logMessages)); - } - public static void warn(String message, Throwable throwable) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.WARN.name(), throwable); - System.out.println(GSON.toJson(logMessages)); - } - - public static void error(String message) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.ERROR.name(), null); - System.out.println(GSON.toJson(logMessages)); + CapaLogEvent event = new CapaLogEvent(CustomLogManager.class.getName(), CapaLogLevel.WARN, message, throwable); + String log = LogAppendManager.buildLog(event); + System.out.println(log); } public static void error(String message, Throwable throwable) { - Map logMessages = LogAppendManager.parseLogs(message, null, CapaLogLevel.ERROR.name(), throwable); - System.out.println(GSON.toJson(logMessages)); + CapaLogEvent event = new CapaLogEvent(CustomLogManager.class.getName(), CapaLogLevel.ERROR, message, throwable); + String log = LogAppendManager.buildLog(event); + System.out.println(log); } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/LogAppendManager.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/LogAppendManager.java index ea5f374..d39acda 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/LogAppendManager.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/manager/LogAppendManager.java @@ -17,13 +17,12 @@ package group.rxcloud.capa.spi.aws.log.manager; import com.google.gson.Gson; -import group.rxcloud.capa.addons.foundation.CapaFoundation; -import group.rxcloud.capa.addons.foundation.FoundationType; import group.rxcloud.capa.component.telemetry.context.CapaContext; -import group.rxcloud.capa.infrastructure.hook.Mixer; +import group.rxcloud.capa.spi.aws.log.appender.CapaLogEvent; import group.rxcloud.capa.spi.aws.log.configuration.CapaComponentLogConfiguration; import group.rxcloud.capa.spi.aws.log.handle.MessageConsumer; import group.rxcloud.capa.spi.aws.log.handle.MessageManager; +import group.rxcloud.capa.spi.aws.log.service.LogMetrics; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.TraceId; import io.opentelemetry.api.trace.Tracer; @@ -35,207 +34,124 @@ import java.time.ZoneId; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; -import java.util.HashMap; import java.util.Map; import java.util.Optional; -public class LogAppendManager { - - /** - * Tag identifier prefix. - */ - protected static final String TAG_PREFIX = "[["; - - /** - * Tag identifier suffix. - */ - protected static final String TAG_SUFFIX = "]]"; +public final class LogAppendManager { /** * The name of log source data. */ - protected static final String LOG_DATA_NAME = "logData"; + private static final String LOG_DATA_NAME = "_log_data"; - protected static final String ERROR_NAME = "errorName"; + private static final String ERROR_NAME = "_error_name"; /** * The name of log level. */ - protected static final String LOG_LEVEL_NAME = "logLevel"; + private static final String LOG_LEVEL_NAME = "_log_level"; /** * The name of logger. */ - protected static final String LOGGER_NAME = "loggerName"; + private static final String LOGGER_NAME = "_logger_name"; /** * The name of thread. */ - protected static final String THREAD_NAME = "threadName"; + private static final String TRACE_ID_NAME = "_trace_id"; /** - * The time of log. + * The name of log's _trace_id. */ - protected static final String LOG_TIME = "logTime"; + private static final String THREAD_NAME = "_thread_name"; /** - * The name of log's _trace_id. + * closure log tag. */ - protected static final String TRACE_ID_NAME = "_trace_id"; - - protected static final String APP_ID_NAME = "appId"; - - protected static final String PUT_LOG_ASYNC_SWITCH = "putLogAsyncSwitch"; - + private static final String LOG_EVENT_NAME = "_log_event"; /** - * Init a {@link Gson} instance. + * closure log tag. */ - private static final Gson GSON = new Gson(); + private static final String THROTTLE = "throttle"; /** - * The namespace for logging error. + * The time of log. */ - private static final String LOG_ERROR_NAMESPACE = "CloudWatchLogs"; - - - private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSZ"); + private static final String LOG_TIME = "_log_time"; - - private static Optional TRACER = Optional.empty(); + private static final String PUT_LOG_ASYNC_SWITCH = "putLogAsyncSwitch"; /** - * Init telemetry hooks and longCounter. + * Init a {@link Gson} instance. */ + private static final Gson GSON = new Gson(); - private static void tryInitTelemetryTracer() { - if (!TRACER.isPresent()) { - try { - Mixer.telemetryHooksNullable().ifPresent(telemetryHooks -> { - TRACER = Optional.ofNullable(telemetryHooks.buildTracer(LOG_ERROR_NAMESPACE).block()); - }); - } catch (Throwable ex) { - CustomLogManager.error("Fail to init telemetry tracer.", ex); - } - } - } + private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSZ"); - protected static Map parseTags(String message, int tagsEndIndex) { - Map tags = null; - int tagStart = 2; - while (tagStart < tagsEndIndex) { - int tagEnd = message.indexOf(',', tagStart); - if (tagEnd < 0 || tagEnd > tagsEndIndex) { - tagEnd = tagsEndIndex; - } - int equalIndex = message.indexOf('=', tagStart); - if (equalIndex > tagStart && equalIndex < tagEnd - 1) { - String key = message.substring(tagStart, equalIndex); - String value = message.substring(equalIndex + 1, tagEnd); - if (tags == null) { - tags = new HashMap<>(); - } - tags.put(key, value); - } - tagStart = tagEnd + 1; - } - return tags; + private LogAppendManager() { } - protected static Map appendMDCTags(Map tags, Map MDCTags) { - if (MDCTags != null && !MDCTags.isEmpty()) { - if (tags == null) { - return new HashMap(MDCTags); - } else { - tags.putAll(MDCTags); - return tags; - } + public static void appendLogs(CapaLogEvent event) { + String logMessage = buildLog(event); + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + if (!configuration.isPresent() + || !configuration.get().containsKey(PUT_LOG_ASYNC_SWITCH) + || Boolean.FALSE.toString().equalsIgnoreCase(configuration.get().get(PUT_LOG_ASYNC_SWITCH))) { + System.out.println(logMessage); + } else { + MessageConsumer consumer = MessageManager.getInstance().getConsumer(); + consumer.processLogEvent(logMessage); } - return tags; } - public static void appendLogs(String message, Map MDCTags, String loggerName, String threadName, - String logLevel, long timestamp, Throwable throwable) { - Map logMessageMap = parseLogs(message, MDCTags, logLevel, throwable); - logMessageMap.put(LOGGER_NAME, loggerName); - logMessageMap.put(THREAD_NAME, threadName); - logMessageMap.put(LOG_TIME, ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()).format(FORMATTER)); + public static String buildLog(CapaLogEvent event) { + Map tags = event.getTags(); + appendDefaultTags(event, tags); // put logs to CloudWatchLogs - if (!logMessageMap.isEmpty()) { - String logMessage = GSON.toJson(logMessageMap); - if (!CapaComponentLogConfiguration.getInstance().containsKey(PUT_LOG_ASYNC_SWITCH) - || Boolean.FALSE.toString().equalsIgnoreCase(CapaComponentLogConfiguration.getInstance().get(PUT_LOG_ASYNC_SWITCH))) { - System.out.println(logMessage); - } else { - MessageConsumer consumer = MessageManager.getInstance().getConsumer(); - consumer.processLogEvent(logMessage); - } - } + return GSON.toJson(tags); } - public static Map parseLogs(String message, Map MDCTags, String logLevel, - Throwable throwable) { - if (StringUtils.isBlank(message)) { - message = ""; - } - Map tags = new HashMap<>(); - if (message.startsWith(TAG_PREFIX)) { - int tagsEndIndex = message.indexOf(TAG_SUFFIX); - if (tagsEndIndex > 0) { - tags = parseTags(message, tagsEndIndex); - if (tags != null) { - message = message.substring(tagsEndIndex + 2); - } + private static void appendDefaultTags(CapaLogEvent event, Map tags) { + // traceId + String traceId = CapaContext.getTraceId(); + if (StringUtils.isNotBlank(traceId) || TraceId.getInvalid().equals(traceId)) { + Optional tracer = LogMetrics.getTracer(); + if (tracer.isPresent()) { + Span span = tracer.get().spanBuilder("CapaLog").startSpan(); + traceId = span.getSpanContext().getTraceId(); + span.end(); } } - tags = appendMDCTags(tags, MDCTags); - Map logMessageMap = new HashMap<>(); - logMessageMap.put(LOG_LEVEL_NAME, logLevel); + if (StringUtils.isNotBlank(traceId)) { + tags.put(TRACE_ID_NAME, traceId); + } + + String message = event.getMessage(); + Throwable throwable = event.getThrowable(); if (throwable != null) { StringWriter sw = new StringWriter(200 * 1024); PrintWriter pw = new PrintWriter(sw); pw.print(message); throwable.printStackTrace(pw); message = sw.toString(); - logMessageMap.put(ERROR_NAME, throwable.getClass().getName()); - } - if (StringUtils.isNotBlank(message)) { - logMessageMap.put(LOG_DATA_NAME, message); + tags.put(ERROR_NAME, throwable.getClass().getName()); } - Map defaultTags = getDefaultTags(); - if (defaultTags != null && !defaultTags.isEmpty()) { - logMessageMap.putAll(defaultTags); - } - if (tags != null && !tags.isEmpty()) { - logMessageMap.putAll(tags); - } - return logMessageMap; - } - protected static Map getDefaultTags() { - Map defaultTags = new HashMap<>(); - // traceId - String traceId = CapaContext.getTraceId(); - if (StringUtils.isNotBlank(traceId) || TraceId.getInvalid().equals(traceId)) { - if (!TRACER.isPresent()) { - tryInitTelemetryTracer(); - } - if (TRACER.isPresent()) { - Span span = TRACER.get().spanBuilder("CapaLog").startSpan(); - traceId = span.getSpanContext().getTraceId(); - span.end(); - } - } - if (StringUtils.isNotBlank(traceId)) { - defaultTags.put(TRACE_ID_NAME, traceId); + if (StringUtils.isNotBlank(message)) { + tags.put(LOG_DATA_NAME, message); } - // appId - String appId = CapaFoundation.getAppId(FoundationType.TRIP); - if (StringUtils.isNotBlank(appId)) { - defaultTags.put(APP_ID_NAME, appId); + tags.put(LOGGER_NAME, event.getLoggerName()); + tags.put(THREAD_NAME, event.getThreadName()); + tags.put(LOG_LEVEL_NAME, event.getLogLevel()); + tags.put(LOG_TIME, + ZonedDateTime.ofInstant(Instant.ofEpochMilli(event.getTime()), ZoneId.systemDefault()) + .format(FORMATTER)); + if (event.isThrottle()) { + tags.put(LOG_EVENT_NAME, THROTTLE); } - return defaultTags; } } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/plugin/AwsLogPlugin.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/plugin/AwsLogPlugin.java deleted file mode 100644 index 60045a8..0000000 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/plugin/AwsLogPlugin.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * 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 group.rxcloud.capa.spi.aws.log.plugin; - -import group.rxcloud.capa.addons.cat.CatLogPlugin; -import group.rxcloud.capa.addons.cat.DefaultCatLogPlugin; -import group.rxcloud.capa.spi.aws.log.configuration.CapaComponentLogConfiguration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Arrays; -import java.util.Map; - -/** - * Aws log plugin. - */ -public class AwsLogPlugin implements CatLogPlugin { - - private static final String KEY_SCENARIO_ENABLE = "scenarioEnable"; - - private static final Logger log = LoggerFactory.getLogger("CapaAwsTagLogger"); - - @Override - public void logTags(String scenario, Map tags) { - if (filterScenario(scenario)) { - log.info(DefaultCatLogPlugin.buildLogData(scenario, tags)); - } - } - - private static boolean filterScenario(String scenario) { - if (!CapaComponentLogConfiguration.getInstance().containsKey(KEY_SCENARIO_ENABLE)) { - return true; - } - - String scenarios = CapaComponentLogConfiguration.getInstance().get(KEY_SCENARIO_ENABLE); - if (scenarios != null) { - String[] splits = scenarios.split(","); - return Arrays.stream(splits) - .anyMatch(s -> s.trim().equalsIgnoreCase(scenario)); - } - return false; - } -} diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/CloudWatchLogsService.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/CloudWatchLogsService.java index 027c2ab..e8ad793 100644 --- a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/CloudWatchLogsService.java +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/CloudWatchLogsService.java @@ -16,8 +16,6 @@ */ package group.rxcloud.capa.spi.aws.log.service; -import group.rxcloud.capa.addons.foundation.CapaFoundation; -import group.rxcloud.capa.addons.foundation.FoundationType; import group.rxcloud.capa.infrastructure.exceptions.CapaException; import group.rxcloud.capa.infrastructure.hook.Mixer; import group.rxcloud.capa.infrastructure.hook.TelemetryHooks; @@ -49,30 +47,44 @@ public class CloudWatchLogsService { private static final CloudWatchLogsClient CLOUD_WATCH_LOGS_CLIENT; + private static final String APP_ID; + private static final String APP_ENV; + private static final String LOG_GROUP_NAME; - private static final String LOG_GROUP_FORMAT = "application/%s/%s"; + + private static final String LOG_GROUP_FORMAT = "/aws/application/%s"; + /** * Log Stream format is appid/count */ - private static final String LOG_STREAM_FORMAT = "%s/%s"; + private static final String LOG_STREAM_FORMAT = "app-%s-cloudwatch-api-%s"; + private static final String CLOUD_WATCH_LOGS_ERROR_NAMESPACE = "CloudWatchLogs"; + private static final String CLOUD_WATCH_LOGS_ERROR_METRIC_NAME = "LogsError"; + private static final String CLOUD_WATCH_LOGS_PUT_LOG_EVENT_ERROR_TYPE = "PutLogEventError"; + private static final String CLOUD_WATCH_LOGS_PUT_LOG_EVENTS_ERROR_TYPE = "PutLogEventsError"; + private static final Integer COUNTER_NUM = 1; private static final Optional TELEMETRY_HOOKS; + private static final int DEFAULT_MAX_LOG_STREAM_COUNT = 10; + private static final List LOG_STREAM_NAMES = new ArrayList<>(); + private static final String LOG_STREAM_COUNT_NAME = "logStreamCount"; + private static Optional LONG_COUNTER = Optional.empty(); static { APP_ENV = buildApplicationEnv(); APP_ID = buildAppId(); - LOG_GROUP_NAME = String.format(LOG_GROUP_FORMAT, APP_ENV, APP_ID); + LOG_GROUP_NAME = String.format(LOG_GROUP_FORMAT, APP_ENV); CLOUD_WATCH_LOGS_CLIENT = CloudWatchLogsClient.builder().build(); createLogGroup(); createLogStream(); @@ -92,7 +104,8 @@ public static void putLogEvent(String message, String logStreamName) { .logGroupName(LOG_GROUP_NAME) .logStreamNamePrefix(logStreamName) .build(); - DescribeLogStreamsResponse describeLogStreamsResponse = CLOUD_WATCH_LOGS_CLIENT.describeLogStreams(logStreamsRequest); + DescribeLogStreamsResponse describeLogStreamsResponse = CLOUD_WATCH_LOGS_CLIENT + .describeLogStreams(logStreamsRequest); String sequenceToken = ""; if (describeLogStreamsResponse != null && !CollectionUtils.isNullOrEmpty(describeLogStreamsResponse.logStreams())) { @@ -120,7 +133,8 @@ public static void putLogEvent(String message, String logStreamName) { try { //Enhance function without affecting function LONG_COUNTER.ifPresent(longCounter -> { - longCounter.bind(Attributes.of(AttributeKey.stringKey(CLOUD_WATCH_LOGS_PUT_LOG_EVENT_ERROR_TYPE), CLOUD_WATCH_LOGS_PUT_LOG_EVENT_ERROR_TYPE)) + longCounter.bind(Attributes.of(AttributeKey.stringKey(CLOUD_WATCH_LOGS_PUT_LOG_EVENT_ERROR_TYPE), + CLOUD_WATCH_LOGS_PUT_LOG_EVENT_ERROR_TYPE)) .add(COUNTER_NUM); }); } finally { @@ -135,7 +149,8 @@ public static void putLogEvents(List messages, String logStreamName) { .logGroupName(LOG_GROUP_NAME) .logStreamNamePrefix(logStreamName) .build(); - DescribeLogStreamsResponse describeLogStreamsResponse = CLOUD_WATCH_LOGS_CLIENT.describeLogStreams(logStreamsRequest); + DescribeLogStreamsResponse describeLogStreamsResponse = CLOUD_WATCH_LOGS_CLIENT + .describeLogStreams(logStreamsRequest); String sequenceToken = ""; if (describeLogStreamsResponse != null && !CollectionUtils.isNullOrEmpty(describeLogStreamsResponse.logStreams())) { @@ -166,7 +181,8 @@ public static void putLogEvents(List messages, String logStreamName) { try { //Enhance function without affecting function LONG_COUNTER.ifPresent(longCounter -> { - longCounter.bind(Attributes.of(AttributeKey.stringKey(CLOUD_WATCH_LOGS_PUT_LOG_EVENTS_ERROR_TYPE), CLOUD_WATCH_LOGS_PUT_LOG_EVENTS_ERROR_TYPE)) + longCounter.bind(Attributes.of(AttributeKey.stringKey(CLOUD_WATCH_LOGS_PUT_LOG_EVENTS_ERROR_TYPE), + CLOUD_WATCH_LOGS_PUT_LOG_EVENTS_ERROR_TYPE)) .add(COUNTER_NUM); }); } finally { @@ -175,26 +191,29 @@ public static void putLogEvents(List messages, String logStreamName) { } private static String buildAppId() { - return CapaFoundation.getAppId(FoundationType.TRIP); + return "TODO"; } private static String buildApplicationEnv() { - return CapaFoundation.getEnv(FoundationType.TRIP) == "" ? "default" : CapaFoundation.getEnv(FoundationType.TRIP).toLowerCase(); + return "TODO"; } private static void createLogGroup() { try { //Describe log group to confirm whether the log group has been created.. DescribeLogGroupsRequest describeLogGroupsRequest = DescribeLogGroupsRequest.builder() - .logGroupNamePrefix(LOG_GROUP_NAME) + .logGroupNamePrefix( + LOG_GROUP_NAME) .build(); - DescribeLogGroupsResponse describeLogGroupsResponse = CLOUD_WATCH_LOGS_CLIENT.describeLogGroups(describeLogGroupsRequest); + DescribeLogGroupsResponse describeLogGroupsResponse = CLOUD_WATCH_LOGS_CLIENT + .describeLogGroups(describeLogGroupsRequest); // If the log group is not created, then create the log group. List logGroups = describeLogGroupsResponse.logGroups(); Boolean hasLogGroup = Boolean.FALSE; if (!CollectionUtils.isNullOrEmpty(logGroups)) { Optional logGroupOptional = logGroups.stream() - .filter(logGroup -> LOG_GROUP_NAME.equalsIgnoreCase(logGroup.logGroupName())) + .filter(logGroup -> LOG_GROUP_NAME + .equalsIgnoreCase(logGroup.logGroupName())) .findAny(); if (logGroupOptional.isPresent()) { hasLogGroup = Boolean.TRUE; @@ -219,16 +238,21 @@ private static void createLogStream() { for (String logStreamName : LOG_STREAM_NAMES) { //Describe log stream to confirm whether the log stream has been created. DescribeLogStreamsRequest describeLogStreamsRequest = DescribeLogStreamsRequest.builder() - .logGroupName(LOG_GROUP_NAME) - .logStreamNamePrefix(logStreamName) + .logGroupName( + LOG_GROUP_NAME) + .logStreamNamePrefix( + logStreamName) .build(); - DescribeLogStreamsResponse describeLogStreamsResponse = CLOUD_WATCH_LOGS_CLIENT.describeLogStreams(describeLogStreamsRequest); + DescribeLogStreamsResponse describeLogStreamsResponse = CLOUD_WATCH_LOGS_CLIENT + .describeLogStreams(describeLogStreamsRequest); // If the log stream is not created, then create the log stream. List logStreams = describeLogStreamsResponse.logStreams(); Boolean hasLogStream = Boolean.FALSE; if (!CollectionUtils.isNullOrEmpty(logStreams)) { Optional logStreamOptional = logStreams.stream() - .filter(logStream -> logStreamName.equalsIgnoreCase(logStream.logStreamName())) + .filter(logStream -> logStreamName + .equalsIgnoreCase( + logStream.logStreamName())) .findAny(); if (logStreamOptional.isPresent()) { hasLogStream = Boolean.TRUE; @@ -250,9 +274,12 @@ private static void createLogStream() { } private static void createLogStreamNames() { - int logStreamCount = CapaComponentLogConfiguration.getInstance().containsKey(LOG_STREAM_COUNT_NAME) - ? Integer.parseInt(LOG_STREAM_COUNT_NAME) - : DEFAULT_MAX_LOG_STREAM_COUNT; + int logStreamCount = DEFAULT_MAX_LOG_STREAM_COUNT; + Optional configuration = CapaComponentLogConfiguration.getInstanceOpt(); + if (configuration.isPresent() && configuration.get().containsKey(LOG_STREAM_COUNT_NAME)) { + logStreamCount = Integer.parseInt(configuration.get().get(LOG_STREAM_COUNT_NAME)); + } + for (int i = 0; i < logStreamCount; i++) { LOG_STREAM_NAMES.add(String.format(LOG_STREAM_FORMAT, APP_ID, i)); } diff --git a/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/LogMetrics.java b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/LogMetrics.java new file mode 100644 index 0000000..1f2df4d --- /dev/null +++ b/capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/service/LogMetrics.java @@ -0,0 +1,107 @@ +/* + * 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 group.rxcloud.capa.spi.aws.log.service; + +import group.rxcloud.capa.infrastructure.hook.Mixer; +import group.rxcloud.capa.spi.aws.log.manager.CustomLogManager; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.Tracer; + +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Log metrics helper. + */ +public final class LogMetrics { + + /** + * The namespace for logging error. + */ + private static final String LOG_NAMESPACE = "Fx.Log"; + + /** + * The metric name for logging error. + */ + private static final String LOG_ERROR_METRIC_NAME = "log_failure_count"; + + /** + * The attribute key for appender type. + */ + private static final String APPENDER_KEY = "appender"; + + /** + * The attribute key for ERROR name. + */ + private static final String ERROR_KEY = "error_name"; + + private static final AtomicBoolean METRIC_INIT = new AtomicBoolean(false); + + /** + * Init an instance of {@link LongCounter}. + */ + private static Optional errorCounter = Optional.empty(); + + private static Optional tracer = Optional.empty(); + + private LogMetrics() { + } + + public static Optional getTracer() { + return tracer; + } + + public static void recordLogError(String appenderName, String errorName) { + try { + getErrorCounter().ifPresent(counter -> { + Attributes attributes = Attributes.builder() + .put(APPENDER_KEY, appenderName) + .put(ERROR_KEY, errorName) + .build(); + counter.add(1, attributes); + }); + } catch (Throwable throwable) { + // ignore any ERROR to keep the log function running. + } + } + + static Optional getErrorCounter() { + if (!METRIC_INIT.get()) { + init(); + } + return errorCounter; + } + + private static void init() { + synchronized (METRIC_INIT) { + if (METRIC_INIT.compareAndSet(false, true)) { + Mixer.telemetryHooksNullable().ifPresent(telemetryHooks -> { + try { + Meter meter = telemetryHooks.buildMeter(LOG_NAMESPACE).block(); + errorCounter = Optional.ofNullable(meter.counterBuilder(LOG_ERROR_METRIC_NAME).build()); + tracer = Optional.ofNullable(telemetryHooks.buildTracer(LOG_NAMESPACE).block()); + } catch (Throwable ex) { + CustomLogManager.warn("Fail to init telemetry components.", ex); + } + }); + } + } + } + +} diff --git a/capa-spi-aws-mesh/pom.xml b/capa-spi-aws-mesh/pom.xml index 4ef4f4d..aeb8a95 100644 --- a/capa-spi-aws-mesh/pom.xml +++ b/capa-spi-aws-mesh/pom.xml @@ -23,7 +23,7 @@ capa-aws-parent group.rxcloud - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE capa-spi-aws-mesh @@ -34,11 +34,6 @@ capa-spi-aws-infrastructure - - group.rxcloud - capa-serializer - - software.amazon.awssdk appmesh diff --git a/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/AwsCapaRpcProperties.java b/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/AwsCapaRpcProperties.java index 71d132e..223240d 100644 --- a/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/AwsCapaRpcProperties.java +++ b/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/AwsCapaRpcProperties.java @@ -16,8 +16,7 @@ */ package group.rxcloud.capa.spi.aws.mesh; -import group.rxcloud.capa.addons.foundation.CapaFoundation; -import group.rxcloud.capa.addons.foundation.FoundationType; +import group.rxcloud.capa.infrastructure.CapaEnvironment; import group.rxcloud.capa.infrastructure.CapaProperties; import group.rxcloud.capa.infrastructure.exceptions.CapaErrorContext; import group.rxcloud.capa.infrastructure.exceptions.CapaException; @@ -62,7 +61,7 @@ abstract class Settings { } // FIXME: 2021/12/15 use trip logic currently - rpcAwsAppMeshNamespace = CapaFoundation.getNamespace(FoundationType.TRIP); + rpcAwsAppMeshNamespace = CapaEnvironment.Provider.getInstance().getDeployEnv(); logger.info("[Capa.Rpc.Client.config] [AwsCapaRpcProperties.AppMeshProperties] rpcAwsAppMeshTemplate[{}] rpcAwsAppMeshPort[{}] rpcAwsAppMeshNamespace[{}]", rpcAwsAppMeshTemplate, rpcAwsAppMeshPort, rpcAwsAppMeshNamespace); diff --git a/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProvider.java b/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProvider.java index 2580062..2d98424 100644 --- a/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProvider.java +++ b/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProvider.java @@ -37,7 +37,11 @@ public interface AwsCapaSerializerProvider { static CapaObjectSerializer getSerializerOrDefault(CapaObjectSerializer originSerializer) { final String serializerName = AwsCapaRpcProperties.SerializerProperties.Settings.getRpcAwsAppMeshSerializer(); Map serializerFactory = AwsCapaSerializerFactory.SERIALIZER_FACTORY; - return serializerFactory.getOrDefault(serializerName, originSerializer); + CapaObjectSerializer configuredSerializer = serializerFactory.get(serializerName); + if (configuredSerializer != null) { + return configuredSerializer; + } + return originSerializer != null ? originSerializer : serializerFactory.get("default"); } /** @@ -51,7 +55,6 @@ final class AwsCapaSerializerFactory { SERIALIZER_FACTORY = new HashMap<>(2, 1); SERIALIZER_FACTORY.put("default", new DefaultObjectSerializer()); - SERIALIZER_FACTORY.put("baiji", new BaijiSSJsonObjectSerializer()); } } } diff --git a/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/BaijiSSJsonObjectSerializer.java b/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/BaijiSSJsonObjectSerializer.java deleted file mode 100644 index 5f4ea52..0000000 --- a/capa-spi-aws-mesh/src/main/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/BaijiSSJsonObjectSerializer.java +++ /dev/null @@ -1,74 +0,0 @@ -/* - * 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 group.rxcloud.capa.spi.aws.mesh.http.serializer; - -import group.rxcloud.capa.addons.serializer.CapaSerializer; -import group.rxcloud.capa.addons.serializer.baiji.ssjson.SSJsonSerializer; -import group.rxcloud.capa.infrastructure.serializer.AbstractObjectSerializer; -import group.rxcloud.cloudruntimes.utils.TypeRef; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.IOException; - -/** - * Serializes and deserializes the object with baiji. - */ -public class BaijiSSJsonObjectSerializer extends AbstractObjectSerializer { - - private static final SSJsonSerializer SERIALIZER = CapaSerializer.ssJsonSerializer; - - /** - * Serializes a given object into byte array. - * - * @param o object to be serialized. - * @return Array of bytes[] with the serialized content. - * @throws IOException In case state cannot be serialized. - */ - @Override - public byte[] doSerialize(Object o) throws IOException { - ByteArrayOutputStream stream = new ByteArrayOutputStream(); - SERIALIZER.serialize(stream, o); - return stream.toByteArray(); - } - - /** - * Deserializes the byte array into the original object. - * - * @param data Content to be parsed. - * @param type Type of the object being deserialized. - * @param Generic type of the object being deserialized. - * @return Object of type T. - * @throws IOException In case content cannot be deserialized. - */ - @Override - public T doDeserialize(byte[] data, TypeRef type) throws IOException { - Class clazz = (Class) type.getType(); - Object deserialize = SERIALIZER.deserialize(new ByteArrayInputStream(data), clazz); - return (T) deserialize; - } - - /** - * Returns the content type of the request. - * - * @return content type of the request - */ - @Override - public String getContentType() { - return SERIALIZER.contentType(); - } -} diff --git a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/AwsCapaHttpTest.java b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/AwsCapaHttpTest.java index 164f3a5..c084ff2 100644 --- a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/AwsCapaHttpTest.java +++ b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/AwsCapaHttpTest.java @@ -18,9 +18,9 @@ import group.rxcloud.capa.component.http.HttpResponse; import group.rxcloud.capa.infrastructure.exceptions.CapaException; +import group.rxcloud.capa.infrastructure.serializer.DefaultObjectSerializer; import group.rxcloud.capa.spi.aws.mesh.http.config.AwsRpcServiceOptions; import group.rxcloud.capa.spi.aws.mesh.http.config.AwsSpiOptionsLoader; -import group.rxcloud.capa.spi.aws.mesh.http.serializer.BaijiSSJsonObjectSerializer; import group.rxcloud.cloudruntimes.utils.TypeRef; import okhttp3.OkHttpClient; import org.junit.jupiter.api.Assertions; @@ -34,17 +34,13 @@ public class AwsCapaHttpTest { private OkHttpClient okHttpClient; - private BaijiSSJsonObjectSerializer baijiSSJsonObjectSerializer; - private AwsCapaHttp awsCapaHttp; @BeforeEach public void setUp() { okHttpClient = new OkHttpClient.Builder().build(); - baijiSSJsonObjectSerializer = new BaijiSSJsonObjectSerializer(); - - awsCapaHttp = new AwsCapaHttp(okHttpClient, baijiSSJsonObjectSerializer); + awsCapaHttp = new AwsCapaHttp(okHttpClient, new DefaultObjectSerializer()); } @Test diff --git a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/config/AwsSpiOptionsLoaderTest.java b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/config/AwsSpiOptionsLoaderTest.java index b2ded10..7189a06 100644 --- a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/config/AwsSpiOptionsLoaderTest.java +++ b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/config/AwsSpiOptionsLoaderTest.java @@ -16,6 +16,7 @@ */ package group.rxcloud.capa.spi.aws.mesh.http.config; +import group.rxcloud.capa.spi.aws.mesh.AwsCapaRpcProperties; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -23,19 +24,33 @@ public class AwsSpiOptionsLoaderTest { @Test public void testLoadRpcServiceOptions_Success() { - System.setProperty("ENV", "meshnamespace"); + String originalEnv = System.getProperty("ENV"); + System.clearProperty("ENV"); - AwsSpiOptionsLoader awsSpiOptionsLoader = new AwsSpiOptionsLoader(); + // Initialize the namespace default before changing the deployment environment. + AwsCapaRpcProperties.AppMeshProperties.Settings.getRpcAwsAppMeshNamespace(); - AwsRpcServiceOptions rpcServiceOptions = awsSpiOptionsLoader.loadRpcServiceOptions("appId"); + try { + System.setProperty("ENV", "meshnamespace"); - Assertions.assertEquals("appId", rpcServiceOptions.getAppId()); + AwsSpiOptionsLoader awsSpiOptionsLoader = new AwsSpiOptionsLoader(); - AwsRpcServiceOptions.AwsToAwsServiceOptions awsToAwsServiceOptions = rpcServiceOptions.getAwsToAwsServiceOptions(); + AwsRpcServiceOptions rpcServiceOptions = awsSpiOptionsLoader.loadRpcServiceOptions("appId"); - Assertions.assertEquals("appId", awsToAwsServiceOptions.getServiceId()); - Assertions.assertEquals(8080, awsToAwsServiceOptions.getServicePort()); - Assertions.assertNotNull(awsToAwsServiceOptions.getServiceEnv()); - Assertions.assertNull(awsToAwsServiceOptions.getNamespace()); + Assertions.assertEquals("appId", rpcServiceOptions.getAppId()); + + AwsRpcServiceOptions.AwsToAwsServiceOptions awsToAwsServiceOptions = rpcServiceOptions.getAwsToAwsServiceOptions(); + + Assertions.assertEquals("appId", awsToAwsServiceOptions.getServiceId()); + Assertions.assertEquals(8080, awsToAwsServiceOptions.getServicePort()); + Assertions.assertEquals("meshnamespace", awsToAwsServiceOptions.getServiceEnv()); + Assertions.assertNull(awsToAwsServiceOptions.getNamespace()); + } finally { + if (originalEnv == null) { + System.clearProperty("ENV"); + } else { + System.setProperty("ENV", originalEnv); + } + } } } diff --git a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProviderTest.java b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProviderTest.java index d8e03da..068bb7c 100644 --- a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProviderTest.java +++ b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/AwsCapaSerializerProviderTest.java @@ -17,14 +17,43 @@ package group.rxcloud.capa.spi.aws.mesh.http.serializer; import group.rxcloud.capa.infrastructure.serializer.CapaObjectSerializer; +import group.rxcloud.capa.infrastructure.serializer.DefaultObjectSerializer; +import group.rxcloud.capa.spi.aws.mesh.AwsCapaRpcProperties; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; public class AwsCapaSerializerProviderTest { + private String originalSerializerSetting; + + @BeforeEach + public void captureSerializerSetting() { + originalSerializerSetting = AwsCapaRpcProperties.SerializerProperties.Settings.getRpcAwsAppMeshSerializer(); + } + + @AfterEach + public void resetSerializerSetting() { + AwsCapaRpcProperties.SerializerProperties.Settings.setRpcAwsAppMeshSerializer(originalSerializerSetting); + } + @Test - public void testGetSerializerOrDefault_Success() { + public void testGetSerializerOrDefault_UsesDefaultWhenConfiguredSerializerIsUnavailable() { + AwsCapaRpcProperties.SerializerProperties.Settings.setRpcAwsAppMeshSerializer("unavailable"); + CapaObjectSerializer serializerOrDefault = AwsCapaSerializerProvider.getSerializerOrDefault(null); + Assertions.assertEquals("application/json", serializerOrDefault.getContentType()); } + + @Test + public void testGetSerializerOrDefault_UsesProvidedSerializer() { + AwsCapaRpcProperties.SerializerProperties.Settings.setRpcAwsAppMeshSerializer("custom"); + CapaObjectSerializer serializer = new DefaultObjectSerializer(); + + CapaObjectSerializer serializerOrDefault = AwsCapaSerializerProvider.getSerializerOrDefault(serializer); + + Assertions.assertSame(serializer, serializerOrDefault); + } } diff --git a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/BaijiSSJsonObjectSerializerTest.java b/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/BaijiSSJsonObjectSerializerTest.java deleted file mode 100644 index cd3a3a1..0000000 --- a/capa-spi-aws-mesh/src/test/java/group/rxcloud/capa/spi/aws/mesh/http/serializer/BaijiSSJsonObjectSerializerTest.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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 group.rxcloud.capa.spi.aws.mesh.http.serializer; - -import group.rxcloud.cloudruntimes.utils.TypeRef; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import java.io.IOException; - -public class BaijiSSJsonObjectSerializerTest { - - private BaijiSSJsonObjectSerializer baijiSSJsonObjectSerializer; - - @BeforeEach - public void setUp() { - baijiSSJsonObjectSerializer = new BaijiSSJsonObjectSerializer(); - } - - @Test - public void testSerializeAndDeserialize_Success() throws IOException { - byte[] serializes = baijiSSJsonObjectSerializer.serialize("serialize"); - - String deserialize = baijiSSJsonObjectSerializer.deserialize(serializes, TypeRef.STRING); - - Assertions.assertEquals("serialize", deserialize); - } - - @Test - public void testGetContentType_Success() { - String contentType = baijiSSJsonObjectSerializer.getContentType(); - Assertions.assertEquals("application/json", contentType); - } -} diff --git a/capa-spi-aws-telemetry/pom.xml b/capa-spi-aws-telemetry/pom.xml index 45507c2..93a1199 100644 --- a/capa-spi-aws-telemetry/pom.xml +++ b/capa-spi-aws-telemetry/pom.xml @@ -23,7 +23,7 @@ capa-aws-parent group.rxcloud - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE capa-spi-aws-telemetry @@ -38,11 +38,6 @@ capa-spi-aws-infrastructure - - group.rxcloud - capa-id-generator - - software.amazon.awssdk diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/AwsCapaTelemetryProperties.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/AwsCapaTelemetryProperties.java index c7830ba..9980eb9 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/AwsCapaTelemetryProperties.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/AwsCapaTelemetryProperties.java @@ -26,19 +26,35 @@ public interface AwsCapaTelemetryProperties { abstract class Settings { private static String awsTraceId = "capa-trace-id"; + private static String defaultMetricNamespce; + private static String[] customizedNamespacePrefix; + private static final String TELEMETRY_AWS_TRACE_ID_KEY = "TELEMETRY_AWS_TRACE_ID_KEY"; + private static final String CUSTOMIZED_METRIC_NAMESPCE_KEY = "CUSTOMIZED_METRIC_NAMESPACE_PREFIX"; + private static final String DEFAULT_METRIC_NAMESPCE = "DEFAULT_METRIC_NAMESPACE"; static { Properties awsProperties = CapaProperties.COMPONENT_PROPERTIES_SUPPLIER.apply("telemetry-aws"); awsTraceId = awsProperties.getProperty(TELEMETRY_AWS_TRACE_ID_KEY, awsTraceId); + customizedNamespacePrefix = awsProperties.getProperty(CUSTOMIZED_METRIC_NAMESPCE_KEY, "") + .split(","); + defaultMetricNamespce = awsProperties.getProperty(DEFAULT_METRIC_NAMESPCE, ""); } public static String getTelemetryAwsTraceIdKey() { return awsTraceId; } + public static String[] getCustomizedNamespacePrefix() { + return customizedNamespacePrefix; + } + + public static String getDefaultMetricNamespce() { + return defaultMetricNamespce; + } + private Settings() { } } diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CloudWatchMetricsExporter.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CloudWatchMetricsExporter.java index e6f94a4..044a68d 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CloudWatchMetricsExporter.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CloudWatchMetricsExporter.java @@ -17,9 +17,9 @@ package group.rxcloud.capa.spi.aws.telemetry.metrics; -import group.rxcloud.capa.addons.foundation.CapaFoundation; -import group.rxcloud.capa.addons.foundation.FoundationType; import group.rxcloud.capa.component.telemetry.SamplerConfig; +import group.rxcloud.capa.infrastructure.CapaEnvironment; +import group.rxcloud.capa.spi.aws.telemetry.AwsCapaTelemetryProperties; import group.rxcloud.capa.spi.telemetry.CapaMetricsExporterSpi; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.sdk.common.Clock; @@ -71,17 +71,19 @@ public class CloudWatchMetricsExporter extends CapaMetricsExporterSpi { private static final MetricsCache METRICS_CACHE = new MetricsCache(); - private static final String APPID = "AppId"; + private static final String APPID = "app_id"; - private static final String ENV = "Env"; + private static final String ENV = "env"; private static final String UNKNOWN = "UNKNOWN"; + private static final String METER = "meter"; + public CloudWatchMetricsExporter(Supplier samplerConfig) { super(samplerConfig); } - private static String getNamespace(MetricData data) { + private static String getMeterName(MetricData data) { return data.getInstrumentationLibraryInfo().getName(); } @@ -91,7 +93,7 @@ private static String getMetricName(MetricData data) { private static String getAppId() { try { - String appId = CapaFoundation.getAppId(FoundationType.TRIP); + String appId = "TODO"; return appId == null ? UNKNOWN : appId; } catch (Throwable e) { return UNKNOWN; @@ -100,14 +102,14 @@ private static String getAppId() { private static String getEnv() { try { - String env = CapaFoundation.getEnv(FoundationType.TRIP); + String env = CapaEnvironment.Provider.getInstance().getDeployEnv(); return env == null ? UNKNOWN : env; } catch (Throwable e) { return UNKNOWN; } } - static List buildDimension(Attributes attributes) { + static List buildDimension(String namespace, String custimizedMeterName, Attributes attributes) { List dimensions = new ArrayList<>(); dimensions.add(Dimension.builder() .name(APPID) @@ -117,6 +119,12 @@ static List buildDimension(Attributes attributes) { .name(ENV) .value(getEnv()) .build()); + if (custimizedMeterName != null && !custimizedMeterName.equals(namespace)) { + dimensions.add(Dimension.builder() + .name(METER) + .value(custimizedMeterName) + .build()); + } if (attributes.isEmpty()) { return dimensions; } @@ -140,64 +148,87 @@ public int compare(Dimension o1, Dimension o2) { } static Map> collectedMetricsByNamespace(Collection metricData) { - Map metricsMap = new HashMap<>(); + Map metricsMap = new HashMap<>(); metricData.forEach(m -> { - String namespace = getNamespace(m); + String meterName = getMeterName(m); String metricName = getMetricName(m); MetricDataType type = m.getType(); if (type == MetricDataType.LONG_SUM) { - processLongPoint(namespace, metricName, metricsMap, m.getLongSumData().getPoints()); + processLongPoint(meterName, metricName, metricsMap, m.getLongSumData().getPoints()); } else if (type == MetricDataType.LONG_GAUGE) { - processLongPoint(namespace, metricName, metricsMap, m.getLongGaugeData().getPoints()); + processLongPoint(meterName, metricName, metricsMap, m.getLongGaugeData().getPoints()); } else if (type == MetricDataType.DOUBLE_SUM) { - processDoublePoint(namespace, metricName, metricsMap, m.getDoubleSumData().getPoints()); + processDoublePoint(meterName, metricName, metricsMap, m.getDoubleSumData().getPoints()); } else if (type == MetricDataType.DOUBLE_GAUGE) { - processDoublePoint(namespace, metricName, metricsMap, m.getDoubleGaugeData().getPoints()); + processDoublePoint(meterName, metricName, metricsMap, m.getDoubleGaugeData().getPoints()); } else if (type == MetricDataType.SUMMARY) { - processDoubleSummary(namespace, metricName, metricsMap, m.getDoubleSummaryData().getPoints()); + processDoubleSummary(meterName, metricName, metricsMap, m.getDoubleSummaryData().getPoints()); } }); Map> metricsMapGroupByNamespace = new HashMap<>(); metricsMap.values() - .forEach(m -> metricsMapGroupByNamespace.computeIfAbsent(m.nameSpace, k -> new ArrayList<>()).add(m)); + .forEach(m -> metricsMapGroupByNamespace.computeIfAbsent(m.namespace, k -> new ArrayList<>()).add(m)); return metricsMapGroupByNamespace; } - static void recordHistogram(String namespace, String metricName, Attributes attributes, double data) { - METRICS_CACHE.recordHistogram(namespace, metricName, attributes, data); + static void recordHistogram(String meterName, String metricName, Attributes attributes, double data) { + METRICS_CACHE.recordHistogram(findNamespace(meterName), meterName, metricName, attributes, data); } - static void recordHistogram(String namespace, String metricName, Attributes attributes, long data) { - METRICS_CACHE.recordHistogram(namespace, metricName, attributes, data); + static void recordHistogram(String meterName, String metricName, Attributes attributes, long data) { + METRICS_CACHE.recordHistogram(findNamespace(meterName), meterName, metricName, attributes, data); } - private static void processLongPoint(String namespace, String metricName, Map metricsMap, + private static void processLongPoint(String meterName, String metricName, Map metricsMap, Collection data) { data.forEach(p -> { long millis = TimeUnit.NANOSECONDS.toMillis(p.getEpochNanos()); + String namespace = findNamespace(meterName); metricsMap.computeIfAbsent( - getKey(namespace, metricName, millis, p.getAttributes()), - k -> new CollectedMetrics(namespace, metricName, millis, buildDimension(p.getAttributes()))) + getKey(namespace, meterName, metricName, millis, p.getAttributes()), + k -> new CollectedMetrics(namespace, meterName, metricName, millis, + buildDimension(namespace, meterName, p.getAttributes()))) .addPoint(BigDecimal.valueOf(p.getValue()).doubleValue()); }); } - private static void processDoublePoint(String namespace, String metricName, - Map metricsMap, Collection data) { + private static String findNamespace(String meterName) { + String[] prefixList = AwsCapaTelemetryProperties.Settings.getCustomizedNamespacePrefix(); + String namespace = meterName; + for (String prefix : prefixList) { + if (!prefix.isEmpty() && namespace.startsWith(prefix)) { + return namespace; + } + } + + // use global namespace if it was set. + String global = AwsCapaTelemetryProperties.Settings.getDefaultMetricNamespce(); + if (global != null && !global.isEmpty()) { + namespace = global; + } + + return namespace; + } + + private static void processDoublePoint(String meterName, String metricName, + Map metricsMap, Collection data) { data.forEach(p -> { long millis = TimeUnit.NANOSECONDS.toMillis(p.getEpochNanos()); - metricsMap.computeIfAbsent(getKey(namespace, metricName, millis, p.getAttributes()), - k -> new CollectedMetrics(namespace, metricName, millis, buildDimension(p.getAttributes()))) + String namespace = findNamespace(meterName); + metricsMap.computeIfAbsent(getKey(namespace, meterName, metricName, millis, p.getAttributes()), + k -> new CollectedMetrics(namespace, meterName, metricName, millis, + buildDimension(namespace, meterName, p.getAttributes()))) .addPoint(p.getValue()); }); } - private static void processDoubleSummary(String namespace, String metricName, - Map metricsMap, + private static void processDoubleSummary(String meterName, String metricName, + Map metricsMap, Collection data) { data.forEach(d -> { long millis = TimeUnit.NANOSECONDS.toMillis(d.getEpochNanos()); + String namespace = findNamespace(meterName); StatisticSet.Builder setBuilder = StatisticSet.builder() .sum(d.getSum()) .sampleCount(BigDecimal.valueOf(d.getCount()).doubleValue()); @@ -210,8 +241,9 @@ private static void processDoubleSummary(String namespace, String metricName, } }); } - metricsMap.computeIfAbsent(getKey(namespace, metricName, millis, d.getAttributes()), - k -> new CollectedMetrics(namespace, metricName, millis, buildDimension(d.getAttributes()))) + metricsMap.computeIfAbsent(getKey(namespace, meterName, metricName, millis, d.getAttributes()), + k -> new CollectedMetrics(namespace, meterName, metricName, millis, + buildDimension(namespace, meterName, d.getAttributes()))) .setStatisticSet(setBuilder.build()); }); } @@ -229,8 +261,9 @@ private static void send(String namespace, List data) { } } - private static String getKey(String nameSpace, String metricName, long epocheMillis, Attributes attributes) { - StringBuilder builder = new StringBuilder(nameSpace + ':' + metricName + ':' + epocheMillis); + private static long getKey(String namespace, String meterName, String metricName, long epocheMillis, + Attributes attributes) { + StringBuilder builder = new StringBuilder(namespace + ':' + meterName + ':' + metricName + ':' + epocheMillis); if (attributes != null && !attributes.isEmpty()) { builder.append(':'); List attrs = new ArrayList<>(); @@ -240,7 +273,7 @@ private static String getKey(String nameSpace, String metricName, long epocheMil attrs.sort(String::compareTo); attrs.forEach(s -> builder.append(s).append('&')); } - return builder.toString(); + return builder.toString().hashCode(); } private static MetricDatum build(CollectedMetrics c, List values, List counts) { @@ -326,7 +359,7 @@ public CompletableResultCode shutdown() { */ static final class MetricsCache { - private final Map[] histogramCache = new ConcurrentHashMap[]{ + private final Map[] histogramCache = new ConcurrentHashMap[]{ new ConcurrentHashMap<>(), new ConcurrentHashMap<>()}; private final AtomicInteger index = new AtomicInteger(); @@ -337,7 +370,7 @@ static final class MetricsCache { MetricsCache() { } - void recordHistogram(String namespace, String metricName, Attributes attributes, T data) { + void recordHistogram(String namespace, String meterName, String metricName, Attributes attributes, T data) { Double value = null; if (data instanceof Double) { value = (Double) data; @@ -359,10 +392,12 @@ void recordHistogram(String namespace, String metricName, Attributes attribu try { long millis = 0L; - histogramCache[currentIndex].computeIfAbsent(getKey(namespace, metricName, millis, attributes), - k -> - new CollectedMetrics(namespace, metricName, millis, buildDimension(attributes))) - .addPoint(value); + histogramCache[currentIndex] + .computeIfAbsent(getKey(namespace, meterName, metricName, millis, attributes), + k -> + new CollectedMetrics(namespace, meterName, metricName, millis, + buildDimension(namespace, meterName, attributes))) + .addPoint(value); } finally { readLock.unlock(); } @@ -372,7 +407,7 @@ void recordHistogram(String namespace, String metricName, Attributes attribu void collectAllByNamespace(Map> result) { synchronized (index) { int currentIndex = changeCache(); - Map cache = histogramCache[currentIndex]; + Map cache = histogramCache[currentIndex]; if (!cache.isEmpty()) { Instant instant = Instant.ofEpochMilli(TimeUnit.NANOSECONDS.toMillis(Clock.getDefault().now())); @@ -382,7 +417,7 @@ void collectAllByNamespace(Map> result) { try { cache.values().forEach(metrics -> { metrics.instant = instant; - result.computeIfAbsent(metrics.nameSpace, key -> new ArrayList<>()).add(metrics); + result.computeIfAbsent(metrics.namespace, key -> new ArrayList<>()).add(metrics); }); cache.clear(); } finally { diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CollectedMetrics.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CollectedMetrics.java index 313007a..9046820 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CollectedMetrics.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/metrics/CollectedMetrics.java @@ -27,7 +27,9 @@ class CollectedMetrics { - String nameSpace; + String namespace; + + String meterName; String metricName; @@ -39,8 +41,9 @@ class CollectedMetrics { StatisticSet statisticSet; - CollectedMetrics(String nameSpace, String metricName, long millis, List dimensions) { - this.nameSpace = nameSpace; + CollectedMetrics(String namespace, String meterName, String metricName, long millis, List dimensions) { + this.namespace = namespace; + this.meterName = meterName; this.metricName = metricName; this.dimensions = dimensions; instant = Instant.ofEpochMilli(millis); diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsTraceIdGenerator.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsTraceIdGenerator.java index 525506c..bcb53b9 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsTraceIdGenerator.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsTraceIdGenerator.java @@ -16,9 +16,10 @@ */ package group.rxcloud.capa.spi.aws.telemetry.trace; -import group.rxcloud.capa.addons.id.generator.TripTraceIdGeneratePolicy; import io.opentelemetry.sdk.trace.IdGenerator; +import java.util.UUID; + /** * Generate trace id form aws trace log. */ @@ -26,11 +27,11 @@ public class AwsTraceIdGenerator implements IdGenerator { @Override public String generateSpanId() { - return TripTraceIdGeneratePolicy.generate(); + return UUID.randomUUID().toString(); } @Override public String generateTraceId() { - return TripTraceIdGeneratePolicy.generate(); + return UUID.randomUUID().toString(); } } diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsWebContextPropagatorLoader.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsWebContextPropagatorLoader.java index 32a2d1f..ac5e8ce 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsWebContextPropagatorLoader.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/AwsWebContextPropagatorLoader.java @@ -16,19 +16,19 @@ */ package group.rxcloud.capa.spi.aws.telemetry.trace; -import com.google.common.collect.Lists; import group.rxcloud.capa.spi.telemetry.ContextPropagatorLoaderSpi; import io.opentelemetry.context.propagation.TextMapPropagator; +import java.util.Collections; import java.util.List; /** - * Load ContextPropagator for ctrip. + * Load ContextPropagator. */ public class AwsWebContextPropagatorLoader extends ContextPropagatorLoaderSpi { @Override public List load() { - return Lists.newArrayList(TraceIdRpcContextPropagator.getInstance()); + return Collections.singletonList(TraceIdRpcContextPropagator.getInstance()); } } diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/LogSpanProcessor.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/LogSpanProcessor.java index 3c55038..41ef003 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/LogSpanProcessor.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/LogSpanProcessor.java @@ -58,7 +58,7 @@ private static StringBuilder addBasicTags(SpanData spanData) { appendTag(builder, "_span_latency_nanos", String.valueOf(spanData.getEndEpochNanos() - spanData.getStartEpochNanos())); if (!spanData.getAttributes().isEmpty()) { spanData.getAttributes().forEach((k, v) -> { - appendTag(builder, "attr." + k.getKey(), String.valueOf(v)); + appendTag(builder, "_attr." + k.getKey(), String.valueOf(v)); }); } builder.append("]]"); diff --git a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/TraceIdRpcContextPropagator.java b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/TraceIdRpcContextPropagator.java index b1da535..01632e1 100644 --- a/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/TraceIdRpcContextPropagator.java +++ b/capa-spi-aws-telemetry/src/main/java/group/rxcloud/capa/spi/aws/telemetry/trace/TraceIdRpcContextPropagator.java @@ -16,7 +16,6 @@ */ package group.rxcloud.capa.spi.aws.telemetry.trace; -import group.rxcloud.capa.addons.id.generator.TripTraceIdGeneratePolicy; import group.rxcloud.capa.component.telemetry.SamplerConfig; import group.rxcloud.capa.component.telemetry.context.CapaContext; import group.rxcloud.capa.spi.aws.telemetry.AwsCapaTelemetryProperties; @@ -33,6 +32,7 @@ import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.UUID; /** * Context Propagator for trip trace id. @@ -73,7 +73,7 @@ public Context extract(Context context, @Nullable C carrier, TextMapGetter capa-aws-parent group.rxcloud - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE 4.0.0 diff --git a/pom.xml b/pom.xml index fdc29f2..2091e86 100644 --- a/pom.xml +++ b/pom.xml @@ -23,10 +23,10 @@ group.rxcloud capa-aws-parent pom - 1.11.13.4-alpha-1 + 1.11.13.5.RELEASE capa-aws-parent AWS for Capa. - https://github.com/reactivegroup + https://github.com/capa-cloud/capa-java-aws @@ -47,9 +47,9 @@ - scm:git:git@github.com:reactivegroup/capa-aws.git - scm:git:git@github.com:reactivegroup/capa-aws.git - git@github.com:reactivegroup/capa-aws.git + scm:git:https://github.com/capa-cloud/capa-java-aws.git + scm:git:git@github.com:capa-cloud/capa-java-aws.git + https://github.com/capa-cloud/capa-java-aws @@ -80,7 +80,6 @@ 1.11.13.2.RELEASE 2.17.40 19.0 - 1.0.9.RELEASE 4.5.13 4.1.68.Final 4.4.11 @@ -109,14 +108,6 @@ import - - group.rxcloud - capa-addons - ${capa-addons.version} - pom - import - - group.rxcloud @@ -670,4 +661,4 @@ - \ No newline at end of file +